diff options
| -rw-r--r-- | srv/default.nix | 2 | ||||
| -rw-r--r-- | srv/src/chat/chat.go | 467 | ||||
| -rw-r--r-- | srv/src/chat/chat_it_test.go | 213 | ||||
| -rw-r--r-- | srv/src/chat/user.go | 68 | ||||
| -rw-r--r-- | srv/src/chat/user_test.go | 26 | ||||
| -rw-r--r-- | srv/src/chat/util.go | 28 | ||||
| -rw-r--r-- | srv/src/cmd/mediocre-blog/main.go | 23 | ||||
| -rw-r--r-- | srv/src/cmd/userid-calc-cli/main.go | 28 | ||||
| -rw-r--r-- | srv/src/http/api.go | 17 | ||||
| -rw-r--r-- | srv/src/http/chat.go | 211 | ||||
| -rw-r--r-- | srv/src/http/tpl/chat.html | 251 |
11 files changed, 4 insertions, 1330 deletions
diff --git a/srv/default.nix b/srv/default.nix index 6bae2a8..95f35c5 100644 --- a/srv/default.nix +++ b/srv/default.nix @@ -38,7 +38,7 @@ pname = "mediocre-blog-srv"; version = "dev"; src = ./src; - vendorSha256 = "sha256-C3hyPDO+6oTUeoGP/ZzBn5Y4V/q1jI12BwkR9NADHn0="; + vendorSha256 = "sha256:0kad8cyg9cd9v7m9l23jf27vkb5yfy9w89xfyrsyj7gd3q0l2yxq"; # disable tests checkPhase = ''''; diff --git a/srv/src/chat/chat.go b/srv/src/chat/chat.go deleted file mode 100644 index 0a88d3b..0000000 --- a/srv/src/chat/chat.go +++ /dev/null @@ -1,467 +0,0 @@ -// Package chat implements a simple chatroom system. -package chat - -import ( - "context" - "encoding/json" - "errors" - "fmt" - "strconv" - "sync" - "time" - - "github.com/mediocregopher/mediocre-go-lib/v2/mctx" - "github.com/mediocregopher/mediocre-go-lib/v2/mlog" - "github.com/mediocregopher/radix/v4" -) - -// ErrInvalidArg is returned from methods in this package when a call fails due -// to invalid input. -type ErrInvalidArg struct { - Err error -} - -func (e ErrInvalidArg) Error() string { - return fmt.Sprintf("invalid argument: %v", e.Err) -} - -var ( - errInvalidMessageID = ErrInvalidArg{Err: errors.New("invalid Message ID")} -) - -// Message describes a message which has been posted to a Room. -type Message struct { - ID string `json:"id"` - UserID UserID `json:"userID"` - Body string `json:"body"` - CreatedAt int64 `json:"createdAt,omitempty"` -} - -func msgFromStreamEntry(entry radix.StreamEntry) (Message, error) { - - // NOTE this should probably be a shortcut in radix - var bodyStr string - for _, field := range entry.Fields { - if field[0] == "json" { - bodyStr = field[1] - break - } - } - - if bodyStr == "" { - return Message{}, errors.New("no 'json' field") - } - - var msg Message - if err := json.Unmarshal([]byte(bodyStr), &msg); err != nil { - return Message{}, fmt.Errorf( - "json unmarshaling body %q: %w", bodyStr, err, - ) - } - - msg.ID = entry.ID.String() - msg.CreatedAt = int64(entry.ID.Time / 1000) - return msg, nil -} - -// MessageIterator returns a sequence of Messages which may or may not be -// unbounded. -type MessageIterator interface { - - // Next blocks until it returns the next Message in the sequence, or the - // context error if the context is cancelled, or io.EOF if the sequence has - // been exhausted. - Next(context.Context) (Message, error) - - // Close should always be called once Next has returned an error or the - // MessageIterator will no longer be used. - Close() error -} - -// HistoryOpts are passed into Room's History method in order to affect its -// result. All fields are optional. -type HistoryOpts struct { - Limit int // defaults to, and is capped at, 100. - Cursor string // If not given then the most recent Messages are returned. -} - -func (o HistoryOpts) sanitize() (HistoryOpts, error) { - if o.Limit <= 0 || o.Limit > 100 { - o.Limit = 100 - } - - if o.Cursor != "" { - id, err := parseStreamEntryID(o.Cursor) - if err != nil { - return HistoryOpts{}, fmt.Errorf("parsing Cursor: %w", err) - } - o.Cursor = id.String() - } - - return o, nil -} - -// Room implements functionality related to a single, unique chat room. -type Room interface { - - // Append accepts a new Message and stores it at the end of the room's - // history. The original Message is returned with any relevant fields (e.g. - // ID) updated. - Append(context.Context, Message) (Message, error) - - // Returns a cursor and the list of historical Messages in time descending - // order. The cursor can be passed into the next call to History to receive - // the next set of Messages. - History(context.Context, HistoryOpts) (string, []Message, error) - - // Listen returns a MessageIterator which will return all Messages appended - // to the Room since the given ID. Once all existing messages are iterated - // through then the MessageIterator will begin blocking until a new Message - // is posted. - Listen(ctx context.Context, sinceID string) (MessageIterator, error) - - // Delete deletes a Message from the Room. - Delete(ctx context.Context, id string) error - - // Close is used to clean up all resources created by the Room. - Close() error -} - -// RoomParams are used to instantiate a new Room. All fields are required unless -// otherwise noted. -type RoomParams struct { - Logger *mlog.Logger - Redis radix.Client - ID string - MaxMessages int -} - -func (p RoomParams) streamKey() string { - return fmt.Sprintf("chat:{%s}:stream", p.ID) -} - -type room struct { - params RoomParams - - closeCtx context.Context - closeCancel context.CancelFunc - wg sync.WaitGroup - - listeningL sync.Mutex - listening map[chan Message]struct{} - listeningLastID radix.StreamEntryID -} - -// NewRoom initializes and returns a new Room instance. -func NewRoom(ctx context.Context, params RoomParams) (Room, error) { - - params.Logger = params.Logger.WithNamespace("chat-room") - - r := &room{ - params: params, - listening: map[chan Message]struct{}{}, - } - - r.closeCtx, r.closeCancel = context.WithCancel(context.Background()) - - // figure out the most recent message, if any. - lastEntryID, err := r.mostRecentMsgID(ctx) - if err != nil { - return nil, fmt.Errorf("discovering most recent entry ID in stream: %w", err) - } - r.listeningLastID = lastEntryID - - r.wg.Add(1) - go func() { - defer r.wg.Done() - r.readStreamLoop(r.closeCtx) - }() - - return r, nil -} - -func (r *room) Close() error { - r.closeCancel() - r.wg.Wait() - return nil -} - -func (r *room) mostRecentMsgID(ctx context.Context) (radix.StreamEntryID, error) { - - var entries []radix.StreamEntry - err := r.params.Redis.Do(ctx, radix.Cmd( - &entries, - "XREVRANGE", r.params.streamKey(), "+", "-", "COUNT", "1", - )) - - if err != nil || len(entries) == 0 { - return radix.StreamEntryID{}, err - } - - return entries[0].ID, nil -} - -func (r *room) Append(ctx context.Context, msg Message) (Message, error) { - msg.ID = "" // just in case - - b, err := json.Marshal(msg) - if err != nil { - return Message{}, fmt.Errorf("json marshaling Message: %w", err) - } - - key := r.params.streamKey() - maxLen := strconv.Itoa(r.params.MaxMessages) - body := string(b) - - var id radix.StreamEntryID - - err = r.params.Redis.Do(ctx, radix.Cmd( - &id, "XADD", key, "MAXLEN", "=", maxLen, "*", "json", body, - )) - - if err != nil { - return Message{}, fmt.Errorf("posting message to redis: %w", err) - } - - msg.ID = id.String() - msg.CreatedAt = int64(id.Time / 1000) - return msg, nil -} - -const zeroCursor = "0-0" - -func (r *room) History(ctx context.Context, opts HistoryOpts) (string, []Message, error) { - opts, err := opts.sanitize() - if err != nil { - return "", nil, err - } - - key := r.params.streamKey() - end := opts.Cursor - if end == "" { - end = "+" - } - start := "-" - count := strconv.Itoa(opts.Limit) - - msgs := make([]Message, 0, opts.Limit) - streamEntries := make([]radix.StreamEntry, 0, opts.Limit) - - err = r.params.Redis.Do(ctx, radix.Cmd( - &streamEntries, - "XREVRANGE", key, end, start, "COUNT", count, - )) - - if err != nil { - return "", nil, fmt.Errorf("calling XREVRANGE: %w", err) - } - - var oldestEntryID radix.StreamEntryID - - for _, entry := range streamEntries { - oldestEntryID = entry.ID - - msg, err := msgFromStreamEntry(entry) - if err != nil { - return "", nil, fmt.Errorf( - "parsing stream entry %q: %w", entry.ID, err, - ) - } - msgs = append(msgs, msg) - } - - if len(msgs) < opts.Limit { - return zeroCursor, msgs, nil - } - - cursor := oldestEntryID.Prev() - return cursor.String(), msgs, nil -} - -func (r *room) readStream(ctx context.Context) error { - - r.listeningL.Lock() - lastEntryID := r.listeningLastID - r.listeningL.Unlock() - - redisAddr := r.params.Redis.Addr() - redisConn, err := radix.Dial(ctx, redisAddr.Network(), redisAddr.String()) - if err != nil { - return fmt.Errorf("creating redis connection: %w", err) - } - defer redisConn.Close() - - streamReader := (radix.StreamReaderConfig{}).New( - redisConn, - map[string]radix.StreamConfig{ - r.params.streamKey(): {After: lastEntryID}, - }, - ) - - for { - dlCtx, dlCtxCancel := context.WithTimeout(ctx, 10*time.Second) - _, streamEntry, err := streamReader.Next(dlCtx) - dlCtxCancel() - - if errors.Is(err, radix.ErrNoStreamEntries) { - continue - } else if err != nil { - return fmt.Errorf("fetching next entry from stream: %w", err) - } - - msg, err := msgFromStreamEntry(streamEntry) - if err != nil { - return fmt.Errorf("parsing stream entry %q: %w", streamEntry, err) - } - - r.listeningL.Lock() - - var dropped int - for ch := range r.listening { - select { - case ch <- msg: - default: - dropped++ - } - } - - if dropped > 0 { - ctx := mctx.Annotate(ctx, "msgID", msg.ID, "dropped", dropped) - r.params.Logger.WarnString(ctx, "some listening channels full, messages dropped") - } - - r.listeningLastID = streamEntry.ID - - r.listeningL.Unlock() - } -} - -func (r *room) readStreamLoop(ctx context.Context) { - for { - err := r.readStream(ctx) - if errors.Is(err, context.Canceled) { - return - } else if err != nil { - r.params.Logger.Error(ctx, "reading from redis stream", err) - } - } -} - -type listenMsgIterator struct { - ch <-chan Message - missedMsgs []Message - sinceEntryID radix.StreamEntryID - cleanup func() -} - -func (i *listenMsgIterator) Next(ctx context.Context) (Message, error) { - - if len(i.missedMsgs) > 0 { - msg := i.missedMsgs[0] - i.missedMsgs = i.missedMsgs[1:] - return msg, nil - } - - for { - select { - case <-ctx.Done(): - return Message{}, ctx.Err() - case msg := <-i.ch: - - entryID, err := parseStreamEntryID(msg.ID) - if err != nil { - return Message{}, fmt.Errorf("parsing Message ID %q: %w", msg.ID, err) - - } else if !i.sinceEntryID.Before(entryID) { - // this can happen if someone Appends a Message at the same time - // as another calls Listen. The Listener might have already seen - // the Message by calling History prior to the stream reader - // having processed it and updating listeningLastID. - continue - } - - return msg, nil - } - } -} - -func (i *listenMsgIterator) Close() error { - i.cleanup() - return nil -} - -func (r *room) Listen( - ctx context.Context, sinceID string, -) ( - MessageIterator, error, -) { - - var sinceEntryID radix.StreamEntryID - - if sinceID != "" { - var err error - if sinceEntryID, err = parseStreamEntryID(sinceID); err != nil { - return nil, fmt.Errorf("parsing sinceID: %w", err) - } - } - - ch := make(chan Message, 32) - - r.listeningL.Lock() - lastEntryID := r.listeningLastID - r.listening[ch] = struct{}{} - r.listeningL.Unlock() - - cleanup := func() { - r.listeningL.Lock() - defer r.listeningL.Unlock() - delete(r.listening, ch) - } - - key := r.params.streamKey() - start := sinceEntryID.Next().String() - end := "+" - if lastEntryID != (radix.StreamEntryID{}) { - end = lastEntryID.String() - } - - var streamEntries []radix.StreamEntry - - err := r.params.Redis.Do(ctx, radix.Cmd( - &streamEntries, - "XRANGE", key, start, end, - )) - - if err != nil { - cleanup() - return nil, fmt.Errorf("retrieving missed stream entries: %w", err) - } - - missedMsgs := make([]Message, len(streamEntries)) - - for i := range streamEntries { - - msg, err := msgFromStreamEntry(streamEntries[i]) - if err != nil { - cleanup() - return nil, fmt.Errorf( - "parsing stream entry %q: %w", streamEntries[i].ID, err, - ) - } - - missedMsgs[i] = msg - } - - return &listenMsgIterator{ - ch: ch, - missedMsgs: missedMsgs, - sinceEntryID: sinceEntryID, - cleanup: cleanup, - }, nil -} - -func (r *room) Delete(ctx context.Context, id string) error { - return r.params.Redis.Do(ctx, radix.Cmd( - nil, "XDEL", r.params.streamKey(), id, - )) -} diff --git a/srv/src/chat/chat_it_test.go b/srv/src/chat/chat_it_test.go deleted file mode 100644 index b0d4431..0000000 --- a/srv/src/chat/chat_it_test.go +++ /dev/null @@ -1,213 +0,0 @@ -//go:build integration -// +build integration - -package chat - -import ( - "context" - "strconv" - "testing" - "time" - - "github.com/google/uuid" - cfgpkg "github.com/mediocregopher/blog.mediocregopher.com/srv/cfg" - "github.com/mediocregopher/mediocre-go-lib/v2/mlog" - "github.com/mediocregopher/radix/v4" - "github.com/stretchr/testify/assert" -) - -const roomTestHarnessMaxMsgs = 10 - -type roomTestHarness struct { - ctx context.Context - room Room - allMsgs []Message -} - -func (h *roomTestHarness) newMsg(t *testing.T) Message { - msg, err := h.room.Append(h.ctx, Message{ - UserID: UserID{ - Name: uuid.New().String(), - Hash: "0000", - }, - Body: uuid.New().String(), - }) - assert.NoError(t, err) - - t.Logf("appended message %s", msg.ID) - - h.allMsgs = append([]Message{msg}, h.allMsgs...) - - if len(h.allMsgs) > roomTestHarnessMaxMsgs { - h.allMsgs = h.allMsgs[:roomTestHarnessMaxMsgs] - } - - return msg -} - -func newRoomTestHarness(t *testing.T) *roomTestHarness { - - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - t.Cleanup(cancel) - - cfg := cfgpkg.NewBlogCfg(cfgpkg.Params{ - Args: []string{}, // prevents the test process args from interfering - }) - - var radixClient cfgpkg.RadixClient - radixClient.SetupCfg(cfg) - t.Cleanup(func() { radixClient.Close() }) - - if err := cfg.Init(ctx); err != nil { - t.Fatal(err) - } - - roomParams := RoomParams{ - Logger: mlog.NewLogger(nil), - Redis: radixClient.Client, - ID: uuid.New().String(), - MaxMessages: roomTestHarnessMaxMsgs, - } - - t.Logf("creating test Room %q", roomParams.ID) - room, err := NewRoom(ctx, roomParams) - assert.NoError(t, err) - - t.Cleanup(func() { - err := radixClient.Client.Do(context.Background(), radix.Cmd( - nil, "DEL", roomParams.streamKey(), - )) - assert.NoError(t, err) - }) - - return &roomTestHarness{ctx: ctx, room: room} -} - -func TestRoom(t *testing.T) { - t.Run("history", func(t *testing.T) { - - tests := []struct { - numMsgs int - limit int - }{ - {numMsgs: 0, limit: 1}, - {numMsgs: 1, limit: 1}, - {numMsgs: 2, limit: 1}, - {numMsgs: 2, limit: 10}, - {numMsgs: 9, limit: 2}, - {numMsgs: 9, limit: 3}, - {numMsgs: 9, limit: 4}, - {numMsgs: 15, limit: 3}, - } - - for i, test := range tests { - t.Run(strconv.Itoa(i), func(t *testing.T) { - t.Logf("test: %+v", test) - - h := newRoomTestHarness(t) - - for j := 0; j < test.numMsgs; j++ { - h.newMsg(t) - } - - var gotMsgs []Message - var cursor string - - for { - - var msgs []Message - var err error - cursor, msgs, err = h.room.History(h.ctx, HistoryOpts{ - Cursor: cursor, - Limit: test.limit, - }) - - assert.NoError(t, err) - assert.NotEmpty(t, cursor) - - if len(msgs) == 0 { - break - } - - gotMsgs = append(gotMsgs, msgs...) - } - - assert.Equal(t, h.allMsgs, gotMsgs) - }) - } - }) - - assertNextMsg := func( - t *testing.T, expMsg Message, - ctx context.Context, it MessageIterator, - ) { - t.Helper() - gotMsg, err := it.Next(ctx) - assert.NoError(t, err) - assert.Equal(t, expMsg, gotMsg) - } - - t.Run("listen/already_populated", func(t *testing.T) { - h := newRoomTestHarness(t) - - msgA, msgB, msgC := h.newMsg(t), h.newMsg(t), h.newMsg(t) - _ = msgA - _ = msgB - - itFoo, err := h.room.Listen(h.ctx, msgC.ID) - assert.NoError(t, err) - defer itFoo.Close() - - itBar, err := h.room.Listen(h.ctx, msgA.ID) - assert.NoError(t, err) - defer itBar.Close() - - msgD := h.newMsg(t) - - // itBar should get msgB and msgC before anything else. - assertNextMsg(t, msgB, h.ctx, itBar) - assertNextMsg(t, msgC, h.ctx, itBar) - - // now both iterators should give msgD - assertNextMsg(t, msgD, h.ctx, itFoo) - assertNextMsg(t, msgD, h.ctx, itBar) - - // timeout should be honored - { - timeoutCtx, timeoutCancel := context.WithTimeout(h.ctx, 1*time.Second) - _, errFoo := itFoo.Next(timeoutCtx) - _, errBar := itBar.Next(timeoutCtx) - timeoutCancel() - - assert.ErrorIs(t, errFoo, context.DeadlineExceeded) - assert.ErrorIs(t, errBar, context.DeadlineExceeded) - } - - // new message should work - { - expMsg := h.newMsg(t) - - timeoutCtx, timeoutCancel := context.WithTimeout(h.ctx, 1*time.Second) - gotFooMsg, errFoo := itFoo.Next(timeoutCtx) - gotBarMsg, errBar := itBar.Next(timeoutCtx) - timeoutCancel() - - assert.Equal(t, expMsg, gotFooMsg) - assert.NoError(t, errFoo) - assert.Equal(t, expMsg, gotBarMsg) - assert.NoError(t, errBar) - } - - }) - - t.Run("listen/empty", func(t *testing.T) { - h := newRoomTestHarness(t) - - it, err := h.room.Listen(h.ctx, "") - assert.NoError(t, err) - defer it.Close() - - msg := h.newMsg(t) - assertNextMsg(t, msg, h.ctx, it) - }) -} diff --git a/srv/src/chat/user.go b/srv/src/chat/user.go deleted file mode 100644 index 3f5ab95..0000000 --- a/srv/src/chat/user.go +++ /dev/null @@ -1,68 +0,0 @@ -package chat - -import ( - "encoding/hex" - "fmt" - "sync" - - "golang.org/x/crypto/argon2" -) - -// UserID uniquely identifies an individual user who has posted a message in a -// Room. -type UserID struct { - - // Name will be the user's chosen display name. - Name string `json:"name"` - - // Hash will be a hex string generated from a secret only the user knows. - Hash string `json:"id"` -} - -// UserIDCalculator is used to calculate UserIDs. -type UserIDCalculator struct { - - // Secret is used when calculating UserID Hash salts. - Secret []byte - - // TimeCost, MemoryCost, and Threads are used as inputs to the Argon2id - // algorithm which is used to generate the Hash. - TimeCost, MemoryCost uint32 - Threads uint8 - - // HashLen specifies the number of bytes the Hash should be. - HashLen uint32 - - // Lock, if set, forces concurrent Calculate calls to occur sequentially. - Lock *sync.Mutex -} - -// NewUserIDCalculator returns a UserIDCalculator with sane defaults. -func NewUserIDCalculator(secret []byte) *UserIDCalculator { - return &UserIDCalculator{ - Secret: secret, - TimeCost: 15, - MemoryCost: 128 * 1024, - Threads: 2, - HashLen: 16, - Lock: new(sync.Mutex), - } -} - -// Calculate accepts a name and password and returns the calculated UserID. -func (c *UserIDCalculator) Calculate(name, password string) UserID { - - input := fmt.Sprintf("%q:%q", name, password) - - hashB := argon2.IDKey( - []byte(input), - c.Secret, // salt - c.TimeCost, c.MemoryCost, c.Threads, - c.HashLen, - ) - - return UserID{ - Name: name, - Hash: hex.EncodeToString(hashB), - } -} diff --git a/srv/src/chat/user_test.go b/srv/src/chat/user_test.go deleted file mode 100644 index 2169cde..0000000 --- a/srv/src/chat/user_test.go +++ /dev/null @@ -1,26 +0,0 @@ -package chat - -import ( - "testing" - - "github.com/stretchr/testify/assert" -) - -func TestUserIDCalculator(t *testing.T) { - - const name, password = "name", "password" - - c := NewUserIDCalculator([]byte("foo")) - - // calculating with same params twice should result in same UserID - userID := c.Calculate(name, password) - assert.Equal(t, userID, c.Calculate(name, password)) - - // changing either name or password should result in a different Hash - assert.NotEqual(t, userID.Hash, c.Calculate(name+"!", password).Hash) - assert.NotEqual(t, userID.Hash, c.Calculate(name, password+"!").Hash) - - // changing the secret should change the UserID - c.Secret = []byte("bar") - assert.NotEqual(t, userID, c.Calculate(name, password)) -} diff --git a/srv/src/chat/util.go b/srv/src/chat/util.go deleted file mode 100644 index 05f4830..0000000 --- a/srv/src/chat/util.go +++ /dev/null @@ -1,28 +0,0 @@ -package chat - -import ( - "strconv" - "strings" - - "github.com/mediocregopher/radix/v4" -) - -func parseStreamEntryID(str string) (radix.StreamEntryID, error) { - - split := strings.SplitN(str, "-", 2) - if len(split) != 2 { - return radix.StreamEntryID{}, errInvalidMessageID - } - - time, err := strconv.ParseUint(split[0], 10, 64) - if err != nil { - return radix.StreamEntryID{}, errInvalidMessageID - } - - seq, err := strconv.ParseUint(split[1], 10, 64) - if err != nil { - return radix.StreamEntryID{}, errInvalidMessageID - } - - return radix.StreamEntryID{Time: time, Seq: seq}, nil -} diff --git a/srv/src/cmd/mediocre-blog/main.go b/srv/src/cmd/mediocre-blog/main.go index 694dd3f..4f8ba78 100644 --- a/srv/src/cmd/mediocre-blog/main.go +++ b/srv/src/cmd/mediocre-blog/main.go @@ -8,7 +8,6 @@ import ( "time" cfgpkg "github.com/mediocregopher/blog.mediocregopher.com/srv/cfg" - "github.com/mediocregopher/blog.mediocregopher.com/srv/chat" "github.com/mediocregopher/blog.mediocregopher.com/srv/http" "github.com/mediocregopher/blog.mediocregopher.com/srv/mailinglist" "github.com/mediocregopher/blog.mediocregopher.com/srv/post" @@ -50,9 +49,6 @@ func main() { defer radixClient.Close() ctx = mctx.WithAnnotator(ctx, &radixClient) - chatGlobalRoomMaxMsgs := cfg.Int("chat-global-room-max-messages", 1000, "Maximum number of messages the global chat room can retain") - chatUserIDCalcSecret := cfg.String("chat-user-id-calc-secret", "", "Secret to use when calculating user ids") - // initialization err := cfg.Init(ctx) @@ -66,10 +62,6 @@ func main() { logger.Fatal(ctx, "initializing", err) } - ctx = mctx.Annotate(ctx, - "chatGlobalRoomMaxMsgs", *chatGlobalRoomMaxMsgs, - ) - clock := clock.Realtime() powStore := pow.NewMemoryStore(clock) @@ -100,19 +92,6 @@ func main() { ml := mailinglist.New(mlParams) - chatGlobalRoom, err := chat.NewRoom(ctx, chat.RoomParams{ - Logger: logger.WithNamespace("global-chat-room"), - Redis: radixClient.Client, - ID: "global", - MaxMessages: *chatGlobalRoomMaxMsgs, - }) - if err != nil { - logger.Fatal(ctx, "initializing global chat room", err) - } - defer chatGlobalRoom.Close() - - chatUserIDCalc := chat.NewUserIDCalculator([]byte(*chatUserIDCalcSecret)) - postSQLDB, err := post.NewSQLDB(dataDir) if err != nil { logger.Fatal(ctx, "initializing sql db for post data", err) @@ -129,8 +108,6 @@ func main() { httpParams.PostAssetStore = postAssetStore httpParams.PostDraftStore = postDraftStore httpParams.MailingList = ml - httpParams.GlobalRoom = chatGlobalRoom - httpParams.UserIDCalculator = chatUserIDCalc logger.Info(ctx, "listening") httpAPI, err := http.New(httpParams) diff --git a/srv/src/cmd/userid-calc-cli/main.go b/srv/src/cmd/userid-calc-cli/main.go deleted file mode 100644 index 90c44e7..0000000 --- a/srv/src/cmd/userid-calc-cli/main.go +++ /dev/null @@ -1,28 +0,0 @@ -package main - -import ( - "encoding/json" - "flag" - "fmt" - - "github.com/mediocregopher/blog.mediocregopher.com/srv/chat" -) - -func main() { - - secret := flag.String("secret", "", "Secret to use when calculating UserIDs") - name := flag.String("name", "", "") - password := flag.String("password", "", "") - - flag.Parse() - - calc := chat.NewUserIDCalculator([]byte(*secret)) - userID := calc.Calculate(*name, *password) - - b, err := json.Marshal(userID) - if err != nil { - panic(err) - } - - fmt.Println(string(b)) -} diff --git a/srv/src/http/api.go b/srv/src/http/api.go index 44b9170..01cad50 100644 --- a/srv/src/http/api.go +++ b/srv/src/http/api.go @@ -17,7 +17,6 @@ import ( lru "github.com/hashicorp/golang-lru" "github.com/mediocregopher/blog.mediocregopher.com/srv/cfg" - "github.com/mediocregopher/blog.mediocregopher.com/srv/chat" "github.com/mediocregopher/blog.mediocregopher.com/srv/http/apiutil" "github.com/mediocregopher/blog.mediocregopher.com/srv/mailinglist" "github.com/mediocregopher/blog.mediocregopher.com/srv/post" @@ -41,9 +40,6 @@ type Params struct { MailingList mailinglist.MailingList - GlobalRoom chat.Room - UserIDCalculator *chat.UserIDCalculator - // PublicURL is the base URL which site visitors can navigate to. PublicURL *url.URL @@ -176,16 +172,9 @@ func (a *api) apiHandler() http.Handler { mux.Handle("/mailinglist/finalize", a.mailingListFinalizeHandler()) mux.Handle("/mailinglist/unsubscribe", a.mailingListUnsubscribeHandler()) - mux.Handle("/chat/global/", http.StripPrefix("/chat/global", newChatHandler( - a.params.GlobalRoom, - a.params.UserIDCalculator, - a.requirePowMiddleware, - ))) - - // disallowGetMiddleware is used rather than a MethodMux because it has an - // exception for websockets, which is needed for chat. - return disallowGetMiddleware(mux) - + return apiutil.MethodMux(map[string]http.Handler{ + "POST": mux, + }) } func (a *api) blogHandler() http.Handler { diff --git a/srv/src/http/chat.go b/srv/src/http/chat.go deleted file mode 100644 index f76e4ad..0000000 --- a/srv/src/http/chat.go +++ /dev/null @@ -1,211 +0,0 @@ -package http - -import ( - "context" - "errors" - "fmt" - "net/http" - "strings" - "unicode" - - "github.com/gorilla/websocket" - "github.com/mediocregopher/blog.mediocregopher.com/srv/chat" - "github.com/mediocregopher/blog.mediocregopher.com/srv/http/apiutil" -) - -type chatHandler struct { - *http.ServeMux - - room chat.Room - userIDCalc *chat.UserIDCalculator - - wsUpgrader websocket.Upgrader -} - -func newChatHandler( - room chat.Room, userIDCalc *chat.UserIDCalculator, - requirePowMiddleware func(http.Handler) http.Handler, -) http.Handler { - c := &chatHandler{ - ServeMux: http.NewServeMux(), - room: room, - userIDCalc: userIDCalc, - - wsUpgrader: websocket.Upgrader{}, - } - - c.Handle("/history", c.historyHandler()) - c.Handle("/user-id", requirePowMiddleware(c.userIDHandler())) - c.Handle("/append", requirePowMiddleware(c.appendHandler())) - c.Handle("/listen", c.listenHandler()) - - return c -} - -func (c *chatHandler) historyHandler() http.Handler { - return http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) { - limit, err := apiutil.StrToInt(r.PostFormValue("limit"), 0) - if err != nil { - apiutil.BadRequest(rw, r, fmt.Errorf("invalid limit parameter: %w", err)) - return - } - - cursor := r.PostFormValue("cursor") - - cursor, msgs, err := c.room.History(r.Context(), chat.HistoryOpts{ - Limit: limit, - Cursor: cursor, - }) - - if argErr := (chat.ErrInvalidArg{}); errors.As(err, &argErr) { - apiutil.BadRequest(rw, r, argErr.Err) - return - } else if err != nil { - apiutil.InternalServerError(rw, r, err) - } - - apiutil.JSONResult(rw, r, struct { - Cursor string `json:"cursor"` - Messages []chat.Message `json:"messages"` - }{ - Cursor: cursor, - Messages: msgs, - }) - }) -} - -func (c *chatHandler) userID(r *http.Request) (chat.UserID, error) { - name := r.PostFormValue("name") - if l := len(name); l == 0 { - return chat.UserID{}, errors.New("name is required") - } else if l > 16 { - return chat.UserID{}, errors.New("name too long") - } - - nameClean := strings.Map(func(r rune) rune { - if !unicode.IsPrint(r) { - return -1 - } - return r - }, name) - - if nameClean != name { - return chat.UserID{}, errors.New("name contains invalid characters") - } - - password := r.PostFormValue("password") - if l := len(password); l == 0 { - return chat.UserID{}, errors.New("password is required") - } else if l > 128 { - return chat.UserID{}, errors.New("password too long") - } - - return c.userIDCalc.Calculate(name, password), nil -} - -func (c *chatHandler) userIDHandler() http.Handler { - return http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) { - userID, err := c.userID(r) - if err != nil { - apiutil.BadRequest(rw, r, err) - return - } - - apiutil.JSONResult(rw, r, struct { - UserID chat.UserID `json:"userID"` - }{ - UserID: userID, - }) - }) -} - -func (c *chatHandler) appendHandler() http.Handler { - return http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) { - userID, err := c.userID(r) - if err != nil { - apiutil.BadRequest(rw, r, err) - return - } - - body := r.PostFormValue("body") - - if l := len(body); l == 0 { - apiutil.BadRequest(rw, r, errors.New("body is required")) - return - - } else if l > 300 { - apiutil.BadRequest(rw, r, errors.New("body too long")) - return - } - - msg, err := c.room.Append(r.Context(), chat.Message{ - UserID: userID, - Body: body, - }) - - if err != nil { - apiutil.InternalServerError(rw, r, err) - return - } - - apiutil.JSONResult(rw, r, struct { - MessageID string `json:"messageID"` - }{ - MessageID: msg.ID, - }) - }) -} - -func (c *chatHandler) listenHandler() http.Handler { - return http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) { - - ctx := r.Context() - sinceID := r.FormValue("sinceID") - - conn, err := c.wsUpgrader.Upgrade(rw, r, nil) - if err != nil { - apiutil.BadRequest(rw, r, err) - return - } - defer conn.Close() - - it, err := c.room.Listen(ctx, sinceID) - - if errors.As(err, new(chat.ErrInvalidArg)) { - apiutil.BadRequest(rw, r, err) - return - - } else if errors.Is(err, context.Canceled) { - return - - } else if err != nil { - apiutil.InternalServerError(rw, r, err) - return - } - - defer it.Close() - - for { - - msg, err := it.Next(ctx) - if errors.Is(err, context.Canceled) { - return - - } else if err != nil { - apiutil.InternalServerError(rw, r, err) - return - } - - err = conn.WriteJSON(struct { - Message chat.Message `json:"message"` - }{ - Message: msg, - }) - - if err != nil { - apiutil.GetRequestLogger(r).Error(ctx, "couldn't write message", err) - return - } - } - }) -} diff --git a/srv/src/http/tpl/chat.html b/srv/src/http/tpl/chat.html deleted file mode 100644 index b2038e2..0000000 --- a/srv/src/http/tpl/chat.html +++ /dev/null @@ -1,251 +0,0 @@ -{{ define "body" }} - -<script async type="module" src="/assets/api.js"></script> - -<style> - #messages { - max-height: 65vh; - overflow: auto; - padding-right: 2rem; - } - - #messages .message { - border: 1px solid #AAA; - border-radius: 10px; - margin-bottom: 1rem; - padding: 2rem; - overflow: auto; - } - - #messages .message .title { - font-weight: bold; - font-size: 120%; - } - - #messages .message .secondaryTitle { - font-family: monospace; - color: #CCC; - } - - #messages .message p { - font-family: monospace; - margin: 1rem 0 0 0; - } - -</style> - -<div id="messages"></div> - -<span id="fail" style="color: red;"></span> - -<script> - -const messagesEl = document.getElementById("messages"); - -let messagesScrolledToBottom = true; -messagesEl.onscroll = () => { - const el = messagesEl; - messagesScrolledToBottom = el.scrollHeight == el.scrollTop + el.clientHeight; -}; - -function renderMessages(msgs) { - - messagesEl.innerHTML = ''; - - msgs.forEach((msg) => { - const el = document.createElement("div"); - el.className = "row message" - - const elWithTextContents = (tag, body) => { - const el = document.createElement(tag); - el.appendChild(document.createTextNode(body)); - return el; - }; - - const titleEl = document.createElement("div"); - titleEl.className = "title"; - el.appendChild(titleEl); - - const userNameEl = elWithTextContents("span", msg.userID.name); - titleEl.appendChild(userNameEl); - - const secondaryTitleEl = document.createElement("div"); - secondaryTitleEl.className = "secondaryTitle"; - el.appendChild(secondaryTitleEl); - - const dt = new Date(msg.createdAt*1000); - const dtStr - = `${dt.getFullYear()}-${dt.getMonth()+1}-${dt.getDate()}` - + ` ${dt.getHours()}:${dt.getMinutes()}:${dt.getSeconds()}`; - - const userIDEl = elWithTextContents("span", `userID:${msg.userID.id} @ ${dtStr}`); - secondaryTitleEl.appendChild(userIDEl); - - const bodyEl = document.createElement("p"); - - const bodyParts = msg.body.split("\n"); - for (const i in bodyParts) { - if (i > 0) bodyEl.appendChild(document.createElement("br")); - bodyEl.appendChild(document.createTextNode(bodyParts[i])); - } - - el.appendChild(bodyEl); - - messagesEl.appendChild(el); - }); -} - - -(async () => { - - const failEl = document.getElementById("fail"); - setErr = (msg) => failEl.innerHTML = `${msg} (please refresh the page to retry)`; - - try { - - const api = await import("/assets/api.js"); - - const history = await api.call("/api/chat/global/history"); - const msgs = history.messages; - - // history returns msgs in time descending, but we display them in time - // ascending. - msgs.reverse() - - const sinceID = (msgs.length > 0) ? msgs[msgs.length-1].id : ""; - - const ws = await api.ws("/api/chat/global/listen", { - params: { sinceID }, - }); - - while (true) { - renderMessages(msgs); - - // If the user was previously scrolled to the bottom then keep them - // there. - if (messagesScrolledToBottom) { - messagesEl.scrollTop = messagesEl.scrollHeight; - } - - const msg = await ws.next(); - msgs.push(msg.message); - renderMessages(msgs); - } - - - } catch (e) { - e = `Failed to fetch message history: ${e}` - setErr(e); - console.error(e); - return; - } - -})() - -</script> - -<style> -#append { - border: 1px dashed #AAA; - border-radius: 10px; - padding: 2rem; -} - -#append #appendBody { - font-family: monospace; -} - -#append #appendStatus { - color: red; -} - -</style> - -<form id="append"> - <h5>New Message</h5> - <div class="row"> - <div class="columns four"> - <input class="u-full-width" placeholder="Name" id="appendName" type="text" /> - <input class="u-full-width" placeholder="Secret" id="appendSecret" type="password" /> - </div> - <div class="columns eight"> - <p> - Your name is displayed alongside your message. - - Your name+secret is used to generate your userID, which is also - displayed alongside your message. - - Other users can validate two messages are from the same person - by comparing the messages' userID. - </p> - </div> - </div> - <div class="row"> - <div class="columns twelve"> - <textarea - style="font-family: monospace" - id="appendBody" - class="u-full-width" - placeholder="Well thought out statement goes here..." - ></textarea> - </div> - </div> - <div class="row"> - <div class="columns four"> - <input class="u-full-width button-primary" id="appendSubmit" type="button" value="Submit" /> - </div> - </div> - <span id="appendStatus"></span> -</form> - -<script> - -const append = document.getElementById("append"); -const appendName = document.getElementById("appendName"); -const appendSecret = document.getElementById("appendSecret"); -const appendBody = document.getElementById("appendBody"); -const appendSubmit = document.getElementById("appendSubmit"); -const appendStatus = document.getElementById("appendStatus"); - -appendSubmit.onclick = async () => { - - const appendSubmitOrigValue = appendSubmit.value; - - appendSubmit.disabled = true; - appendSubmit.className = ""; - appendSubmit.value = "Please hold..."; - - appendStatus.innerHTML = ''; - - try { - - const api = await import("/assets/api.js"); - - await api.call('/api/chat/global/append', { - body: { - name: appendName.value, - password: appendSecret.value, - body: appendBody.value, - }, - requiresPow: true, - }); - - appendBody.value = ''; - - } catch (e) { - - appendStatus.innerHTML = e; - - } finally { - appendSubmit.disabled = false; - appendSubmit.className = "button-primary"; - appendSubmit.value = appendSubmitOrigValue; - } -}; - -</script> - -{{ end }} - -{{ template "base.html" . }} - |