Register the broker wait before the first poll (closes #157)

GET /api/v1/messages read the client output queue and only then
registered with the broker. Notify wakes only the channels registered
at that moment, so a message queued between the empty read and the
registration waited out the whole long-poll timeout. The handler now
registers first and removes its channel on every exit path; the IRC
relay loop registers before each drain in the same way.

The handlers read the queue through a function field that defaults to
Database.PollMessages, so the new test can queue a message just after
the first read finds nothing and check that it comes back at once.

Model: opus-5-5

internal/handlers/api_test.go

@@ -140,6 +140,7 @@ type testServer struct {
 	fxApp      *fxtest.App
 	broker     *broker.Broker
 	database   *db.Database
+	handlers   *handlers.Handlers
 }
 
 func newTestServer(
@@ -170,6 +171,7 @@ func newTestServerWith(
 		srv      *server.Server
 		brk      *broker.Broker
 		database *db.Database
+		hdlr     *handlers.Handlers
 	)
 
 	app := fxtest.New(t,
@@ -210,7 +212,7 @@ func newTestServerWith(
 			newTestHandlers,
 			newTestServerFx,
 		),
-		fx.Populate(&srv, &brk, &database),
+		fx.Populate(&srv, &brk, &database, &hdlr),
 	)
 
 	app.RequireStart()
@@ -228,6 +230,7 @@ func newTestServerWith(
 		fxApp:      app,
 		broker:     brk,
 		database:   database,
+		handlers:   hdlr,
 	}
 }
 
@@ -2554,6 +2557,110 @@ func TestLongPollTimeout(t *testing.T) {
 	}
 }
 
+// TestLongPollReturnsMessageQueuedAfterFirstRead checks that
+// a message queued, and the broker notified, after the long
+// poll's first read of the queue found nothing but before the
+// poll waits, is returned at once and not when the poll times
+// out.
+func TestLongPollReturnsMessageQueuedAfterFirstRead(t *testing.T) {
+	const (
+		nick   = "lp_late"
+		sender = "lp_late_sender"
+
+		// pollTimeout is the long poll's timeout, in seconds.
+		pollTimeout = 10
+
+		// atOnce is the longest the long poll may take to
+		// return the message.
+		atOnce = 5 * time.Second
+	)
+
+	tserver := newTestServer(t)
+	token := tserver.createSession(nick)
+
+	_, lastID := tserver.pollMessages(token, 0)
+
+	sessionID, err := tserver.database.GetSessionByNick(t.Context(), nick)
+	if err != nil {
+		t.Fatalf("get session of %s: %v", nick, err)
+	}
+
+	queued := make(chan error, 1)
+
+	var once sync.Once
+
+	tserver.handlers.SetPollMessages(func(
+		ctx context.Context,
+		clientID, afterQueueID int64,
+		limit int,
+	) ([]db.IRCMessage, int64, error) {
+		msgs, lastQID, pollErr := tserver.database.PollMessages(
+			ctx, clientID, afterQueueID, limit,
+		)
+
+		once.Do(func() {
+			queued <- tserver.queueMessage(ctx, sessionID, sender, nick)
+		})
+
+		return msgs, lastQID, pollErr
+	})
+
+	start := time.Now()
+
+	msgs, err := tserver.longPoll(token, lastID, pollTimeout)
+	if err != nil {
+		t.Fatalf("long poll: %v", err)
+	}
+
+	elapsed := time.Since(start)
+
+	select {
+	case queueErr := <-queued:
+		if queueErr != nil {
+			t.Fatalf("queue message: %v", queueErr)
+		}
+	default:
+		t.Fatal("the long poll did not read the queue")
+	}
+
+	if elapsed > atOnce {
+		t.Fatalf("the long poll returned after %v, want within %v",
+			elapsed, atOnce)
+	}
+
+	if !findMessage(msgs, privmsgCmd, sender) {
+		t.Fatalf("long poll: got %v, want the PRIVMSG from %s",
+			msgs, sender)
+	}
+}
+
+// queueMessage stores a PRIVMSG from sender to nick, queues
+// it to the session sessionID and notifies the broker, as the
+// server does for a DM. It returns errors instead of failing
+// the test, so a handler's goroutine can call it.
+func (tserver *testServer) queueMessage(
+	ctx context.Context,
+	sessionID int64,
+	sender, nick string,
+) error {
+	msgID, _, err := tserver.database.InsertMessage(
+		ctx, privmsgCmd, sender, nick, sessionID, 0,
+		nil, json.RawMessage(`["hello"]`), nil,
+	)
+	if err != nil {
+		return fmt.Errorf("insert message: %w", err)
+	}
+
+	err = tserver.database.EnqueueToSession(ctx, sessionID, msgID)
+	if err != nil {
+		return fmt.Errorf("enqueue: %w", err)
+	}
+
+	tserver.broker.Notify(sessionID)
+
+	return nil
+}
+
 func TestEphemeralChannelCleanup(t *testing.T) {
 	const channel = "#ephemeral"