feat(create-room): DM, botDM, and channel creation across sites

inbox-worker/handler.go

@@ -3,12 +3,17 @@ package main
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"log/slog"
+	"strings"
 	"time"
 
+	"go.mongodb.org/mongo-driver/v2/mongo"
+
 	"github.com/hmchangw/chat/pkg/idgen"
 	"github.com/hmchangw/chat/pkg/model"
+	"github.com/hmchangw/chat/pkg/natsutil"
 )
 
 // InboxStore abstracts the data store operations needed by the inbox worker.
@@ -50,6 +55,8 @@ func (h *Handler) HandleEvent(ctx context.Context, data []byte) error {
 		return h.handleRoleUpdated(ctx, &evt)
 	case "thread_subscription_upserted":
 		return h.handleThreadSubscriptionUpserted(ctx, &evt)
+	case model.MessageTypeRoomCreated:
+		return h.handleRoomCreated(ctx, &evt)
 	default:
 		slog.Warn("unknown event type, skipping", "type", evt.Type)
 		return nil
@@ -97,6 +104,7 @@ func (h *Handler) handleMemberAdded(ctx context.Context, evt *model.OutboxEvent)
 			RoomType:           model.RoomTypeChannel,
 			SiteID:             event.SiteID,
 			Roles:              []model.Role{model.RoleMember},
+			Name:               event.RoomName,
 			HistorySharedSince: historySharedSince,
 			JoinedAt:           joinedAt,
 		}
@@ -182,3 +190,101 @@ func (h *Handler) handleThreadSubscriptionUpserted(ctx context.Context, evt *mod
 	}
 	return nil
 }
+
+// errPermanent signals a non-retryable error; callers should Ack and move on.
+var errPermanent = errors.New("permanent")
+
+func rolesForType(t model.RoomType) []model.Role {
+	if t == model.RoomTypeChannel {
+		return []model.Role{model.RoleMember}
+	}
+	return nil
+}
+
+func subscriptionName(d *model.RoomCreatedOutbox, u *model.User) string {
+	switch d.RoomType {
+	case model.RoomTypeChannel, model.RoomTypeDiscussion:
+		return d.RoomName
+	case model.RoomTypeDM, model.RoomTypeBotDM:
+		// On the remote site, the "other party" relative to u is the requester.
+		return d.RequesterAccount
+	}
+	return ""
+}
+
+// isBot mirrors the bot predicate used by room-service/helper.go and pkg/pipelines:
+// accounts ending in ".bot" or starting with "p_" (webhook-style bots).
+func isBot(account string) bool {
+	return strings.HasSuffix(account, ".bot") || strings.HasPrefix(account, "p_")
+}
+
+func subscriptionIsSubscribed(d *model.RoomCreatedOutbox, u *model.User) bool {
+	if d.RoomType != model.RoomTypeBotDM {
+		return false
+	}
+	return !isBot(u.Account)
+}
+
+func (h *Handler) handleRoomCreated(ctx context.Context, evt *model.OutboxEvent) error {
+	requestID := natsutil.RequestIDFromContext(ctx)
+	if requestID == "" {
+		return fmt.Errorf("missing X-Request-ID: %w", errPermanent)
+	}
+
+	var data model.RoomCreatedOutbox
+	if err := json.Unmarshal(evt.Payload, &data); err != nil {
+		return fmt.Errorf("unmarshal room_created payload: %w: %w", err, errPermanent)
+	}
+	if len(data.Accounts) == 0 {
+		slog.Warn("room_created event with empty Accounts list",
+			"requestId", requestID, "roomId", data.RoomID)
+		return nil
+	}
+
+	users, err := h.store.FindUsersByAccounts(ctx, data.Accounts)
+	if err != nil {
+		return fmt.Errorf("find users by accounts: %w", err)
+	}
+	// FindUsersByAccounts can return a subset; treat any account in
+	// data.Accounts that didn't come back as a hard failure rather than
+	// silently materializing partial remote-side state with no retry signal.
+	userByAccount := make(map[string]model.User, len(users))
+	for i := range users {
+		userByAccount[users[i].Account] = users[i]
+	}
+	for _, account := range data.Accounts {
+		if _, ok := userByAccount[account]; !ok {
+			return fmt.Errorf("find users by accounts: missing account %q (room %s home %s)",
+				account, data.RoomID, data.HomeSiteID)
+		}
+	}
+
+	acceptedAt := time.UnixMilli(data.Timestamp).UTC()
+	subs := make([]*model.Subscription, 0, len(data.Accounts))
+	for _, account := range data.Accounts {
+		u := userByAccount[account]
+		sub := &model.Subscription{
+			ID:           idgen.GenerateUUIDv7(),
+			User:         model.SubscriptionUser{ID: u.ID, Account: u.Account},
+			RoomID:       data.RoomID,
+			SiteID:       data.HomeSiteID,
+			Roles:        rolesForType(data.RoomType),
+			Name:         subscriptionName(&data, &u),
+			RoomType:     data.RoomType,
+			IsSubscribed: subscriptionIsSubscribed(&data, &u),
+			JoinedAt:     acceptedAt,
+		}
+		subs = append(subs, sub)
+	}
+
+	if len(subs) == 0 {
+		return nil
+	}
+	if err := h.store.BulkCreateSubscriptions(ctx, subs); err != nil {
+		if mongo.IsDuplicateKeyError(err) {
+			return nil
+		}
+		return fmt.Errorf("bulk create subs: %w", err)
+	}
+	return nil
+}

inbox-worker/handler_test.go

@@ -11,7 +11,9 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/hmchangw/chat/pkg/idgen"
 	"github.com/hmchangw/chat/pkg/model"
+	"github.com/hmchangw/chat/pkg/natsutil"
 )
 
 // --- In-memory InboxStore stub ---
@@ -911,3 +913,204 @@ type errorThreadSubStore struct {
 func (s *errorThreadSubStore) UpsertThreadSubscription(_ context.Context, _ *model.ThreadSubscription) error {
 	return fmt.Errorf("boom")
 }
+
+func TestRolesForType(t *testing.T) {
+	assert.Equal(t, []model.Role{model.RoleMember}, rolesForType(model.RoomTypeChannel))
+	assert.Nil(t, rolesForType(model.RoomTypeDM))
+	assert.Nil(t, rolesForType(model.RoomTypeBotDM))
+}
+
+func TestSubscriptionName(t *testing.T) {
+	d := model.RoomCreatedOutbox{
+		RoomType:         model.RoomTypeChannel,
+		RoomName:         "deal team",
+		RequesterAccount: "alice",
+	}
+	assert.Equal(t, "deal team", subscriptionName(&d, &model.User{Account: "bob"}))
+
+	d.RoomType = model.RoomTypeDM
+	assert.Equal(t, "alice", subscriptionName(&d, &model.User{Account: "bob"}))
+
+	d.RoomType = model.RoomTypeBotDM
+	assert.Equal(t, "alice", subscriptionName(&d, &model.User{Account: "weather.bot"}))
+}
+
+func TestSubscriptionIsSubscribed(t *testing.T) {
+	d := model.RoomCreatedOutbox{RoomType: model.RoomTypeChannel}
+	assert.False(t, subscriptionIsSubscribed(&d, &model.User{Account: "bob"}))
+
+	d.RoomType = model.RoomTypeDM
+	assert.False(t, subscriptionIsSubscribed(&d, &model.User{Account: "bob"}))
+
+	d.RoomType = model.RoomTypeBotDM
+	assert.False(t, subscriptionIsSubscribed(&d, &model.User{Account: "weather.bot"}))
+	assert.True(t, subscriptionIsSubscribed(&d, &model.User{Account: "alice"}))
+	// p_ webhook bots: same as .bot — bot side gets IsSubscribed=false.
+	assert.False(t, subscriptionIsSubscribed(&d, &model.User{Account: "p_webhook"}))
+}
+
+func TestHandleRoomCreatedRequiresRequestID(t *testing.T) {
+	store := &stubInboxStore{}
+	h := NewHandler(store)
+	payload, _ := json.Marshal(model.RoomCreatedOutbox{
+		RoomID: "r1", RoomType: model.RoomTypeChannel,
+		Accounts: []string{"bob"},
+	})
+	err := h.handleRoomCreated(context.Background(), &model.OutboxEvent{Payload: payload})
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "missing X-Request-ID")
+}
+
+func TestHandleRoomCreatedEmptyAccountsAcksWithWarn(t *testing.T) {
+	store := &stubInboxStore{}
+	h := NewHandler(store)
+	const reqID = "0193abcd-0193-7abc-89ab-0193abcd0193"
+	ctx := natsutil.WithRequestID(context.Background(), reqID)
+
+	payload, _ := json.Marshal(model.RoomCreatedOutbox{
+		RoomID: "r1", RoomType: model.RoomTypeChannel, Accounts: []string{},
+	})
+	require.NoError(t, h.handleRoomCreated(ctx, &model.OutboxEvent{Payload: payload}))
+}
+
+func TestHandleRoomCreatedDMBuildsRemoteSub(t *testing.T) {
+	store := &stubInboxStore{
+		users: []model.User{
+			{ID: "u_bob", Account: "bob", SiteID: "site-B"},
+		},
+	}
+	h := NewHandler(store)
+	const reqID = "0193abcd-0193-7abc-89ab-0193abcd0193"
+	ctx := natsutil.WithRequestID(context.Background(), reqID)
+
+	payload, _ := json.Marshal(model.RoomCreatedOutbox{
+		RoomID:           "u_aliceu_bob",
+		RoomType:         model.RoomTypeDM,
+		RoomName:         "",
+		HomeSiteID:       "site-A",
+		Accounts:         []string{"bob"},
+		RequesterAccount: "alice",
+		Timestamp:        1740000000000,
+	})
+	require.NoError(t, h.handleRoomCreated(ctx, &model.OutboxEvent{Payload: payload}))
+
+	subs := store.bulkSubscriptions
+	require.Len(t, subs, 1)
+	assert.True(t, idgen.IsValidUUIDv7(subs[0].ID))
+	assert.Equal(t, "u_aliceu_bob", subs[0].RoomID)
+	assert.Equal(t, "site-A", subs[0].SiteID)
+	assert.Equal(t, "alice", subs[0].Name)
+	assert.Nil(t, subs[0].Roles)
+	assert.False(t, subs[0].IsSubscribed)
+	assert.Equal(t, model.RoomTypeDM, subs[0].RoomType)
+}
+
+func TestHandleRoomCreatedChannelBulkInsert(t *testing.T) {
+	store := &stubInboxStore{
+		users: []model.User{
+			{ID: "u_bob", Account: "bob", SiteID: "site-B"},
+			{ID: "u_ian", Account: "ian", SiteID: "site-B"},
+		},
+	}
+	h := NewHandler(store)
+	const reqID = "0193abcd-0193-7abc-89ab-0193abcd0193"
+	ctx := natsutil.WithRequestID(context.Background(), reqID)
+
+	payload, _ := json.Marshal(model.RoomCreatedOutbox{
+		RoomID:           "r1",
+		RoomType:         model.RoomTypeChannel,
+		RoomName:         "deal team",
+		HomeSiteID:       "site-A",
+		Accounts:         []string{"bob", "ian"},
+		RequesterAccount: "alice",
+		Timestamp:        1,
+	})
+	require.NoError(t, h.handleRoomCreated(ctx, &model.OutboxEvent{Payload: payload}))
+
+	subs := store.bulkSubscriptions
+	require.Len(t, subs, 2)
+	for _, s := range subs {
+		assert.Equal(t, "deal team", s.Name)
+		assert.Equal(t, []model.Role{model.RoleMember}, s.Roles)
+		assert.Equal(t, model.RoomTypeChannel, s.RoomType)
+		assert.Equal(t, "site-A", s.SiteID)
+	}
+}
+
+func TestHandleMemberAddedSetsNameAndRoomType(t *testing.T) {
+	store := &stubInboxStore{
+		users: []model.User{
+			{ID: "u_bob", Account: "bob", SiteID: "site-B"},
+		},
+	}
+	h := NewHandler(store)
+
+	change := model.MemberAddEvent{
+		Type:      "member_added",
+		RoomID:    "r1",
+		RoomName:  "deal team",
+		Accounts:  []string{"bob"},
+		SiteID:    "site-A",
+		JoinedAt:  1740000000000,
+		Timestamp: 1740000000000,
+	}
+	changeData, err := json.Marshal(change)
+	require.NoError(t, err)
+
+	evt := model.OutboxEvent{
+		Type:       "member_added",
+		SiteID:     "site-A",
+		DestSiteID: "site-B",
+		Payload:    changeData,
+	}
+	evtData, err := json.Marshal(evt)
+	require.NoError(t, err)
+
+	require.NoError(t, h.HandleEvent(context.Background(), evtData))
+
+	subs := store.getSubscriptions()
+	require.Len(t, subs, 1)
+	assert.Equal(t, "deal team", subs[0].Name)
+	assert.Equal(t, model.RoomTypeChannel, subs[0].RoomType)
+}
+
+func TestHandleRoomCreatedBotDMBuildsRemoteBotSub(t *testing.T) {
+	// Cross-site botDM: human (alice) is the requester on site-A; bot
+	// (weather.bot) lives on site-B. The outbox event lands at site-B's
+	// inbox-worker, which must materialize the bot's sub with:
+	//   Name        = human's account ("alice")
+	//   IsSubscribed = false
+	//   Roles       = nil (no member role for botDM)
+	//   SiteID      = home site (site-A)
+	store := &stubInboxStore{
+		users: []model.User{
+			{ID: "u_weather", Account: "weather.bot", SiteID: "site-B"},
+		},
+	}
+	h := NewHandler(store)
+	const reqID = "0193abcd-0193-7abc-89ab-0193abcd0193"
+	ctx := natsutil.WithRequestID(context.Background(), reqID)
+
+	payload, _ := json.Marshal(model.RoomCreatedOutbox{
+		RoomID:           "u_aliceu_weather",
+		RoomType:         model.RoomTypeBotDM,
+		RoomName:         "",
+		HomeSiteID:       "site-A",
+		Accounts:         []string{"weather.bot"},
+		RequesterAccount: "alice",
+		Timestamp:        1740000000000,
+	})
+	require.NoError(t, h.handleRoomCreated(ctx, &model.OutboxEvent{Payload: payload}))
+
+	subs := store.bulkSubscriptions
+	require.Len(t, subs, 1, "exactly one remote sub for the bot")
+	assert.True(t, idgen.IsValidUUIDv7(subs[0].ID))
+	assert.Equal(t, "u_aliceu_weather", subs[0].RoomID)
+	assert.Equal(t, "site-A", subs[0].SiteID, "bot's sub.siteID is the room's home site")
+	assert.Equal(t, "alice", subs[0].Name, "bot's sub.Name is the human account")
+	assert.Nil(t, subs[0].Roles)
+	assert.False(t, subs[0].IsSubscribed)
+	assert.Equal(t, model.RoomTypeBotDM, subs[0].RoomType)
+	assert.Equal(t, "u_weather", subs[0].User.ID)
+	assert.Equal(t, "weather.bot", subs[0].User.Account)
+}