feat: spotlight-org sync from hr-syncer events

go.mod

@@ -17,6 +17,7 @@ require (
 	github.com/hashicorp/vault/api v1.23.0
 	github.com/hashicorp/vault/api/auth/approle v0.12.0
 	github.com/hashicorp/vault/api/auth/kubernetes v0.12.0
+	github.com/klauspost/compress v1.18.5
 	github.com/minio/minio-go/v7 v7.1.0
 	github.com/nats-io/jwt/v2 v2.8.1
 	github.com/nats-io/nats-server/v2 v2.12.6
@@ -100,7 +101,6 @@ require (
 	github.com/hashicorp/go-sockaddr v1.0.7 // indirect
 	github.com/hashicorp/hcl v1.0.1-vault-7 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
-	github.com/klauspost/compress v1.18.5 // indirect
 	github.com/klauspost/cpuid/v2 v2.3.0 // indirect
 	github.com/klauspost/crc32 v1.3.0 // indirect
 	github.com/leodido/go-urn v1.4.0 // indirect

pkg/stream/stream.go

@@ -80,3 +80,13 @@ func MigrationOplog(siteID string) Config {
 		Subjects: []string{subject.MigrationOplogWildcard(siteID)},
 	}
 }
+
+// OrgSyncStream is the HR_{centralSiteID} stream populated by hr-syncer's
+// daily publishes on chat.hr.{centralSiteID}.>. hr-syncer runs at one
+// central site; every fab site's search-sync-worker consumes from it.
+func OrgSyncStream(centralSiteID string) Config {
+	return Config{
+		Name:     fmt.Sprintf("HR_%s", centralSiteID),
+		Subjects: []string{fmt.Sprintf("chat.hr.%s.>", centralSiteID)},
+	}
+}

pkg/stream/stream_test.go

@@ -25,6 +25,7 @@ func TestStreamConfigs(t *testing.T) {
 		{"MessagesCanonical", stream.MessagesCanonical(siteID), "MESSAGES_CANONICAL_site-a", "chat.msg.canonical.site-a.>"},
 		{"Rooms", stream.Rooms(siteID), "ROOMS_site-a", "chat.room.canonical.site-a.>"},
 		{"PushNotification", stream.PushNotification(siteID), "PUSH_NOTIFICATION_site-a", "chat.server.notification.push.site-a.>"},
+		{"OrgSyncStream", stream.OrgSyncStream(siteID), "HR_site-a", "chat.hr.site-a.>"},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {

pkg/subject/subject.go

@@ -1146,3 +1146,10 @@ func MigrationOplog(siteID, collection, op string) string {
 func MigrationOplogWildcard(siteID string) string {
 	return fmt.Sprintf("chat.oplog.%s.>", siteID)
 }
+
+// OrgSyncEmployeesUpsert is the subject search-sync-worker's spotlight-org
+// collection consumes from; hr-syncer publishes on the same subject at the
+// central site.
+func OrgSyncEmployeesUpsert(centralSiteID string) string {
+	return fmt.Sprintf("chat.hr.%s.employees.upsert", centralSiteID)
+}

pkg/subject/subject_test.go

@@ -914,3 +914,8 @@ func TestTeamsSubjectBuilders(t *testing.T) {
 		assert.Equal(t, "chat.user.{account}.request.teams.site-a.call.user", subject.TeamsUserCallPattern("site-a"))
 	})
 }
+
+func TestOrgSyncEmployeesUpsert(t *testing.T) {
+	got := subject.OrgSyncEmployeesUpsert("site-a")
+	assert.Equal(t, "chat.hr.site-a.employees.upsert", got)
+}

search-sync-worker/collection.go

@@ -33,8 +33,8 @@ type Collection interface {
 	// emit lightweight `{"script":{"id":...}}` references instead of repeating
 	// the full source in every fan-out bulk action.
 	StoredScripts() map[string]json.RawMessage
-	// BuildAction converts raw JetStream message data into one or more
-	// BulkActions. An empty slice means the event should be acked without
-	// any ES write (e.g., filtered out).
+	// BuildAction converts the already-decompressed JetStream message body
+	// into one or more BulkActions (empty slice = ack with no ES write,
+	// e.g. filtered).
 	BuildAction(data []byte) ([]searchengine.BulkAction, error)
 }

search-sync-worker/compression.go

@@ -0,0 +1,53 @@
+package main
+
+import (
+	"fmt"
+
+	"github.com/klauspost/compress/zstd"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// Compression is signalled by NATS headers on the incoming message.
+// Currently only zstd is supported; unknown values fall through to the
+// raw body.
+const (
+	headerNatsEncoding = "Nats-Encoding"
+	encodingZstd       = "zstd"
+
+	// maxDecompressedBytes caps zstd output size at decode time via
+	// WithDecoderMaxMemory. A hostile or corrupt frame that decodes to
+	// more than this is aborted mid-decode rather than allowed to grow
+	// the destination buffer to arbitrary size.
+	maxDecompressedBytes = 16 * 1024 * 1024
+)
+
+// zstdDecoder is a process-wide decoder reused across every message.
+// klauspost's decoder is safe for concurrent DecodeAll calls, and
+// creating one per message pays a decoder-init cost per JetStream
+// delivery — a real allocation on the consumer hot path.
+var zstdDecoder = mustNewZstdDecoder()
+
+func mustNewZstdDecoder() *zstd.Decoder {
+	d, err := zstd.NewReader(nil,
+		zstd.WithDecoderConcurrency(1),
+		zstd.IgnoreChecksum(true),
+		zstd.WithDecoderMaxMemory(maxDecompressedBytes),
+	)
+	if err != nil {
+		panic(fmt.Sprintf("init zstd decoder: %v", err))
+	}
+	return d
+}
+
+// decodePayload decompresses the body when Nats-Encoding signals a
+// supported codec, otherwise returns it unchanged.
+func decodePayload(msg jetstream.Msg) ([]byte, error) {
+	if msg.Headers().Get(headerNatsEncoding) != encodingZstd {
+		return msg.Data(), nil
+	}
+	out, err := zstdDecoder.DecodeAll(msg.Data(), nil)
+	if err != nil {
+		return nil, fmt.Errorf("decompress zstd payload: %w", err)
+	}
+	return out, nil
+}

search-sync-worker/deploy/docker-compose.yml

@@ -16,6 +16,10 @@ services:
       - SEARCH_BACKEND=elasticsearch
       - MSG_INDEX_PREFIX=messages-site-local-v1
       - SPOTLIGHT_INDEX=spotlight-site-local-v1
+      - SPOTLIGHT_ORG_INDEX=spotlightorg-site-local-v1
+      # hr-syncer runs at one central site; for single-site local dev,
+      # point at ourselves.
+      - HR_CENTRAL_SITE_ID=site-local
       - USER_ROOM_INDEX=user-room-mv-site-local
       - BOOTSTRAP_STREAMS=true
       # Optional YYYY-MM-DD (UTC) cutoff for the messages collection.

search-sync-worker/handler.go

@@ -60,11 +60,17 @@ func NewHandler(store Store, collection Collection, bulkSize int) *Handler {
 	}
 }
 
-// Add parses a JetStream message via the collection and adds its actions to
-// the buffer. If the collection produces zero actions (e.g., a filtered
-// event), the message is immediately acked without touching the buffer.
+// Add parses a JetStream message via the collection and adds its actions
+// to the buffer. If the collection produces zero actions, the message is
+// acked without touching the buffer.
 func (h *Handler) Add(msg jetstream.Msg) {
-	actions, err := h.collection.BuildAction(msg.Data())
+	data, err := decodePayload(msg)
+	if err != nil {
+		slog.Error("decode payload", "error", err)
+		natsutil.Ack(msg, "decode payload failed")
+		return
+	}
+	actions, err := h.collection.BuildAction(data)
 	if err != nil {
 		slog.Error("build action", "error", err)
 		natsutil.Ack(msg, "build action failed")

search-sync-worker/handler_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/klauspost/compress/zstd"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 	"github.com/stretchr/testify/assert"
@@ -19,9 +20,10 @@ import (
 
 // stubMsg implements jetstream.Msg for testing.
 type stubMsg struct {
-	data   []byte
-	acked  bool
-	nacked bool
+	data    []byte
+	headers nats.Header
+	acked   bool
+	nacked  bool
 }
 
 func (m *stubMsg) Data() []byte                              { return m.data }
@@ -34,7 +36,7 @@ func (m *stubMsg) TermWithReason(string) error               { return nil }
 func (m *stubMsg) Metadata() (*jetstream.MsgMetadata, error) { return nil, nil }
 func (m *stubMsg) Subject() string                           { return "" }
 func (m *stubMsg) Reply() string                             { return "" }
-func (m *stubMsg) Headers() nats.Header                      { return nil }
+func (m *stubMsg) Headers() nats.Header                      { return m.headers }
 func (m *stubMsg) DoubleAck(context.Context) error           { return nil }
 
 func makeStubMsg(t *testing.T, evt *model.MessageEvent) *stubMsg {
@@ -47,7 +49,7 @@ func makeStubMsg(t *testing.T, evt *model.MessageEvent) *stubMsg {
 func TestHandler_Add(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	store := NewMockStore(ctrl)
-	h := NewHandler(store, newMessageCollection("msgs-v1", time.Time{}), 500)
+	h := NewHandler(store, newMessageCollection("msgs-v1", time.Time{}, false), 500)
 
 	evt := model.MessageEvent{
 		Event: model.EventCreated,
@@ -66,7 +68,7 @@ func TestHandler_Add(t *testing.T) {
 func TestHandler_Add_MalformedJSON(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	store := NewMockStore(ctrl)
-	h := NewHandler(store, newMessageCollection("msgs-v1", time.Time{}), 500)
+	h := NewHandler(store, newMessageCollection("msgs-v1", time.Time{}, false), 500)
 
 	msg := &stubMsg{data: []byte("{invalid")}
 	h.Add(msg)
@@ -92,7 +94,7 @@ func TestHandler_Flush(t *testing.T) {
 			Bulk(gomock.Any(), gomock.Len(1)).
 			Return([]searchengine.BulkResult{{Status: 201}}, nil)
 
-		h := NewHandler(store, newMessageCollection("msgs-v1", time.Time{}), 500)
+		h := NewHandler(store, newMessageCollection("msgs-v1", time.Time{}, false), 500)
 		msg := makeStubMsg(t, &baseEvt)
 		h.Add(msg)
 		h.Flush(context.Background())
@@ -109,7 +111,7 @@ func TestHandler_Flush(t *testing.T) {
 			Bulk(gomock.Any(), gomock.Len(1)).
 			Return([]searchengine.BulkResult{{Status: 409, Error: "version conflict"}}, nil)
 
-		h := NewHandler(store, newMessageCollection("msgs-v1", time.Time{}), 500)
+		h := NewHandler(store, newMessageCollection("msgs-v1", time.Time{}, false), 500)
 		msg := makeStubMsg(t, &baseEvt)
 		h.Add(msg)
 		h.Flush(context.Background())
@@ -125,7 +127,7 @@ func TestHandler_Flush(t *testing.T) {
 			Bulk(gomock.Any(), gomock.Len(1)).
 			Return([]searchengine.BulkResult{{Status: 500, Error: "internal"}}, nil)
 
-		h := NewHandler(store, newMessageCollection("msgs-v1", time.Time{}), 500)
+		h := NewHandler(store, newMessageCollection("msgs-v1", time.Time{}, false), 500)
 		msg := makeStubMsg(t, &baseEvt)
 		h.Add(msg)
 		h.Flush(context.Background())
@@ -141,7 +143,7 @@ func TestHandler_Flush(t *testing.T) {
 			Bulk(gomock.Any(), gomock.Len(2)).
 			Return(nil, fmt.Errorf("connection refused"))
 
-		h := NewHandler(store, newMessageCollection("msgs-v1", time.Time{}), 500)
+		h := NewHandler(store, newMessageCollection("msgs-v1", time.Time{}, false), 500)
 		msg1 := makeStubMsg(t, &baseEvt)
 		evt2 := baseEvt
 		evt2.Message.ID = "m2"
@@ -159,7 +161,7 @@ func TestHandler_Flush(t *testing.T) {
 	t.Run("empty flush is no-op", func(t *testing.T) {
 		ctrl := gomock.NewController(t)
 		store := NewMockStore(ctrl)
-		h := NewHandler(store, newMessageCollection("msgs-v1", time.Time{}), 500)
+		h := NewHandler(store, newMessageCollection("msgs-v1", time.Time{}, false), 500)
 		h.Flush(context.Background())
 		assert.Equal(t, 0, h.MessageCount())
 	})
@@ -175,7 +177,7 @@ func TestHandler_Flush(t *testing.T) {
 				{Status: 500, Error: "shard failure"},
 			}, nil)
 
-		h := NewHandler(store, newMessageCollection("msgs-v1", time.Time{}), 500)
+		h := NewHandler(store, newMessageCollection("msgs-v1", time.Time{}, false), 500)
 		msgs := make([]*stubMsg, 3)
 		for i := range msgs {
 			evt := baseEvt
@@ -502,3 +504,55 @@ func TestHandler_FanOut(t *testing.T) {
 		assert.True(t, msg1.nacked, "msg 1 had one failing action → nakked")
 	})
 }
+
+// TestHandler_ZstdDecompression exercises the Nats-Encoding:zstd path
+// via a spy Collection that captures the bytes BuildAction receives.
+// Ensures compression is applied transparently at the handler layer so
+// collections never see the compressed frame.
+func TestHandler_ZstdDecompression(t *testing.T) {
+	original := []byte(`{"timestamp":1,"employees":[{"sectId":"S1"}]}`)
+	enc, err := zstd.NewWriter(nil, zstd.WithEncoderLevel(zstd.SpeedDefault))
+	require.NoError(t, err)
+	compressed := enc.EncodeAll(original, nil)
+	require.NoError(t, enc.Close())
+
+	spy := &captureCollection{}
+	h := NewHandler(NewMockStore(gomock.NewController(t)), spy, 10)
+	h.Add(&stubMsg{
+		data:    compressed,
+		headers: nats.Header{"Nats-Encoding": []string{"zstd"}},
+	})
+
+	assert.Equal(t, original, spy.received, "BuildAction should receive decompressed bytes")
+}
+
+// TestHandler_UncompressedPassthrough — plain messages with no
+// Nats-Encoding header reach BuildAction unchanged.
+func TestHandler_UncompressedPassthrough(t *testing.T) {
+	payload := []byte(`{"timestamp":1,"employees":[]}`)
+
+	spy := &captureCollection{}
+	h := NewHandler(NewMockStore(gomock.NewController(t)), spy, 10)
+	h.Add(&stubMsg{data: payload})
+
+	assert.Equal(t, payload, spy.received)
+}
+
+// captureCollection records the bytes passed to BuildAction so tests
+// can assert on the exact payload the handler forwarded.
+type captureCollection struct {
+	received []byte
+}
+
+func (c *captureCollection) StreamConfig(string) jetstream.StreamConfig {
+	return jetstream.StreamConfig{}
+}
+func (c *captureCollection) ConsumerName() string                      { return "capture" }
+func (c *captureCollection) FilterSubjects(string) []string            { return nil }
+func (c *captureCollection) TemplateName() string                      { return "" }
+func (c *captureCollection) TemplateBody() json.RawMessage             { return nil }
+func (c *captureCollection) StoredScripts() map[string]json.RawMessage { return nil }
+func (c *captureCollection) BuildAction(data []byte) ([]searchengine.BulkAction, error) {
+	c.received = append(c.received[:0], data...)
+	return nil, nil
+}

search-sync-worker/inbox_integration_test.go

@@ -157,8 +157,8 @@ func TestSpotlightSync_Integration(t *testing.T) {
 	require.NoError(t, err)
 	waitForClusterGreen(t, esURL, 120*time.Second)
 
-	coll := newSpotlightCollection(indexName)
-	require.NoError(t, engine.UpsertTemplate(ctx, coll.TemplateName(), overrideIndexSettings(spotlightTemplateBody(indexName))))
+	coll := newSpotlightCollection(indexName, true)
+	require.NoError(t, engine.UpsertTemplate(ctx, coll.TemplateName(), overrideIndexSettings(coll.TemplateBody())))
 	preCreateIndex(t, esURL, indexName)
 	waitForClusterGreen(t, esURL, 120*time.Second)
 
@@ -249,8 +249,8 @@ func TestSpotlightSync_BulkInvite(t *testing.T) {
 	require.NoError(t, err)
 	waitForClusterGreen(t, esURL, 120*time.Second)
 
-	coll := newSpotlightCollection(indexName)
-	require.NoError(t, engine.UpsertTemplate(ctx, coll.TemplateName(), overrideIndexSettings(spotlightTemplateBody(indexName))))
+	coll := newSpotlightCollection(indexName, true)
+	require.NoError(t, engine.UpsertTemplate(ctx, coll.TemplateName(), overrideIndexSettings(coll.TemplateBody())))
 	preCreateIndex(t, esURL, indexName)
 	waitForClusterGreen(t, esURL, 120*time.Second)