feat: spotlight-org sync from hr-syncer events#170
Merged
Conversation
📝 WalkthroughWalkthroughThis PR adds a new ChangesSpotlight Org Sync Implementation
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Possibly related PRs
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
🧹 Nitpick comments (2)
search-sync-worker/template_test.go (1)
22-35: 💤 Low valueConsider guarding type assertions with
require.NotNilchecks.The type assertions on lines 25-34 will panic if
customAnalyzerSettings()returns an unexpected structure. While test panics are acceptable failures, addingrequire.NotNilchecks before each type assertion would make failures clearer and provide better diagnostics.🛡️ Example: safer assertions
func TestCustomAnalyzerSettings_Shape(t *testing.T) { got := customAnalyzerSettings() analyzer := got["analyzer"].(map[string]any) + require.NotNil(t, analyzer, "analyzer section must be present") custom := analyzer["custom_analyzer"].(map[string]any) + require.NotNil(t, custom, "custom_analyzer must be present") assert.Equal(t, "custom", custom["type"]) ...🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@search-sync-worker/template_test.go` around lines 22 - 35, TestCustomAnalyzerSettings_Shape uses unchecked type assertions on the structure returned by customAnalyzerSettings(), which can panic and obscure failures; before each assertion, add require.NotNil checks for got, analyzer := got["analyzer"], custom := analyzer["custom_analyzer"], tokenizer := got["tokenizer"], and tok := tokenizer["custom_tokenizer"] (or equivalent intermediate values) to ensure those keys exist and are non-nil, then proceed with the type assertions and existing assert.Equal checks in the TestCustomAnalyzerSettings_Shape function.search-sync-worker/spotlight_org.go (1)
135-142: 💤 Low valueOptional: surface
gzip.Reader.Close()error instead of silently discarding viadefer.
gr.Close()can return a checksum/truncation error after the body has already been read. Withdefer gr.Close()the error is dropped, so a corrupt-tail payload silently parses as a successful (possibly truncated) JSON. Low-impact in practice —io.ReadAlltypically surfaces the same error first — but worth tightening since this is the trust boundary for an external publisher.♻️ Suggested rewrite
func gunzipBytes(b []byte) ([]byte, error) { gr, err := gzip.NewReader(bytes.NewReader(b)) if err != nil { return nil, err } - defer gr.Close() - return io.ReadAll(gr) + data, readErr := io.ReadAll(gr) + closeErr := gr.Close() + if readErr != nil { + return nil, readErr + } + if closeErr != nil { + return nil, closeErr + } + return data, nil }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@search-sync-worker/spotlight_org.go` around lines 135 - 142, The current gunzipBytes function defers gzip.NewReader.Close(), which discards any Close() error (e.g., checksum/truncation) — change the function to explicitly call io.ReadAll(gr) into a variable, then call gr.Close() and if Close() returns a non-nil error return that error (or if both Read and Close produce errors prefer the read error but surface the close error when present); update the implementation around gzip.NewReader, io.ReadAll and gr.Close() so Close() errors are not silently ignored.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@search-sync-worker/spotlight_org.go`:
- Around line 135-142: The current gunzipBytes function defers
gzip.NewReader.Close(), which discards any Close() error (e.g.,
checksum/truncation) — change the function to explicitly call io.ReadAll(gr)
into a variable, then call gr.Close() and if Close() returns a non-nil error
return that error (or if both Read and Close produce errors prefer the read
error but surface the close error when present); update the implementation
around gzip.NewReader, io.ReadAll and gr.Close() so Close() errors are not
silently ignored.
In `@search-sync-worker/template_test.go`:
- Around line 22-35: TestCustomAnalyzerSettings_Shape uses unchecked type
assertions on the structure returned by customAnalyzerSettings(), which can
panic and obscure failures; before each assertion, add require.NotNil checks for
got, analyzer := got["analyzer"], custom := analyzer["custom_analyzer"],
tokenizer := got["tokenizer"], and tok := tokenizer["custom_tokenizer"] (or
equivalent intermediate values) to ensure those keys exist and are non-nil, then
proceed with the type assertions and existing assert.Equal checks in the
TestCustomAnalyzerSettings_Shape function.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 5e7e4d29-64c0-4fdf-a305-4b2423ad6386
📒 Files selected for processing (23)
docs/superpowers/plans/2026-05-11-spotlight-org-sync.mddocs/superpowers/specs/2026-05-11-spotlight-org-sync-design.mdpkg/model/hrsync.gopkg/model/model_test.gopkg/stream/stream.gopkg/stream/stream_test.gopkg/subject/subject.gopkg/subject/subject_test.gosearch-sync-worker/deploy/docker-compose.ymlsearch-sync-worker/handler_test.gosearch-sync-worker/inbox_integration_test.gosearch-sync-worker/integration_test.gosearch-sync-worker/main.gosearch-sync-worker/messages.gosearch-sync-worker/messages_test.gosearch-sync-worker/spotlight.gosearch-sync-worker/spotlight_org.gosearch-sync-worker/spotlight_org_test.gosearch-sync-worker/spotlight_test.gosearch-sync-worker/template.gosearch-sync-worker/template_test.gosearch-sync-worker/user_room.gosearch-sync-worker/user_room_test.go
Joey0538
pushed a commit
that referenced
this pull request
May 11, 2026
gzip.Reader.Close reports trailing checksum / truncation errors that io.ReadAll can miss on a corrupted stream. Replace the deferred close with an explicit one so a truncated publisher payload doesn't parse as a successful (short) JSON. Per CodeRabbit review on PR #170. https://claude.ai/code/session_01QLFefxiCHDP24LDLQzLLG2
Joey0538
force-pushed
the
claude/spotlight-org-sync-design-fS2qh
branch
from
98c3e80 to
e1618c4
Compare
Joey0538
force-pushed
the
claude/spotlight-org-sync-design-fS2qh
branch
from
56b7b8f to
019f4d8
Compare
mliu33
reviewed
Jun 30, 2026
| // publishes on hr.sync.{siteID}.>. Schema is owned by hr-syncer. | ||
| func HRSync(siteID string) Config { | ||
| return Config{ | ||
| Name: fmt.Sprintf("HR_SYNC_%s", siteID), |
Collaborator
There was a problem hiding this comment.
This is the old naming and subject names. It has been updated, plz check the latest one
| } | ||
|
|
||
| func HRSyncEmployeesUpsert(siteID string) string { | ||
| return fmt.Sprintf("hr.sync.%s.employees.upsert", siteID) |
| } | ||
|
|
||
| func HRSyncUsersUpsert(siteID string) string { | ||
| return fmt.Sprintf("hr.sync.%s.users.upsert", siteID) |
| // search-sync-worker is a pure consumer of both and must not create | ||
| // their schemas. | ||
| inboxName := stream.Inbox(cfg.SiteID).Name | ||
| hrSyncName := stream.HRSync(cfg.SiteID).Name |
Collaborator
There was a problem hiding this comment.
It has been updated. Plz check latest one
mliu33
reviewed
Jul 1, 2026
| DeptTCName string `json:"deptTCName,omitempty" es:"search_as_you_type,custom_analyzer"` | ||
| DeptName string `json:"deptName,omitempty" es:"search_as_you_type,custom_analyzer"` | ||
| DeptDescription string `json:"deptDescription,omitempty" es:"search_as_you_type,custom_analyzer"` | ||
| DivisionID string `json:"divisionId,omitempty" es:"search_as_you_type,custom_analyzer"` |
Collaborator
There was a problem hiding this comment.
Do we support searching division id in current behavior ?
Collaborator
Author
There was a problem hiding this comment.
No, but this data was still present in ES
Joey0538
force-pushed
the
claude/spotlight-org-sync-design-fS2qh
branch
3 times, most recently
from
8592ce1 to
58f1ee3
Compare
- stream.HR(siteID) → HR_{siteID} stream carrying chat.hr.{siteID}.>
publishes. Schema is owned by hr-syncer.
- subject.HREmployeesUpsert / HRUsersUpsert build the corresponding
employees / users subjects.
https://claude.ai/code/session_01QLFefxiCHDP24LDLQzLLG2
… templates Add a devMode field to messageCollection and spotlightCollection. Each constructor takes devMode bool and threads it through TemplateBody to *TemplateBody(prefix, devMode). Shard/replica counts are computed inline: messages prod 4/2 -> dev 1/0 spotlight prod 3/1 -> dev 1/0 Consumer wires cfg.DevMode into both new-collection callsites in main.go (applied in the spotlight-org commit alongside the new collection's own devMode plumbing). user-room is untouched. https://claude.ai/code/session_01QLFefxiCHDP24LDLQzLLG2
…employees.upsert
Maintains a per-section ES index (spotlightorg-{siteID}-vN) keyed by
sectId, sourced from hr-syncer's batched HR account publishes on the
new HR stream.
Design highlights:
- SpotlightOrgIndex is one struct serving three roles: wire-side row
unmarshal target, ES doc body on write, and source of truth for the
ES mapping via esPropertiesFromStruct. Nine string fields with
omitempty and search_as_you_type / custom_analyzer ES mapping.
- Local hrSyncEmployeeBatch{Timestamp, Employees} type in
search-sync-worker (NOT pkg/model) — the full Employee struct lives
in hr-syncer's internal repo, and JSON unmarshal into
[]SpotlightOrgIndex reads only the org fields we care about, ignoring
the rest.
- Collection.BuildAction interface signature changes from ([]byte) to
(jetstream.Msg) so BuildAction implementations can read NATS message
headers. All existing collections updated to the new signature.
- Compression is signalled by the Nats-Encoding: zstd header (matching
hr-syncer's convention). When present, the message body is a raw
zstd frame decompressed with a 16MB size cap to prevent OOM on
hostile payloads.
- BuildAction dedupes by SectID keeping last-wins per batch and emits
one ES _update per unique sectId with doc_as_upsert:true. Doc-merge +
omitempty means partial-field publishes preserve untouched stored
fields — no painless script needed. Empty SectID rows and empty
batches return (nil, nil) so the handler acks with no ES write.
- ActionUpdate is used WITHOUT external versioning; handler.go's 409
logic for ActionUpdate depends on Version=0.
- StoredScripts() returns nil - the collection uses no inline scripts.
- devMode wired into all three new-collection callsites in main.go
(messages, spotlight, spotlight-org). Includes SPOTLIGHT_ORG_INDEX
as a required env var (vN-suffixed like other search indices) and
the HR bootstrap skip (same as INBOX ownership pattern).
Integration test publishes zstd-compressed batches over NATS with the
Nats-Encoding header and verifies doc upsert + doc-merge field
preservation across subsequent updates.
Also carries: docker-compose SPOTLIGHT_ORG_INDEX + DEV_MODE entries;
klauspost/compress upgraded from indirect to direct dependency;
design + implementation-plan docs.
https://claude.ai/code/session_01QLFefxiCHDP24LDLQzLLG2
Joey0538
force-pushed
the
claude/spotlight-org-sync-design-fS2qh
branch
from
58f1ee3 to
0114f16
Compare
mliu33
approved these changes
Jul 1, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
spotlight-orgcollection insearch-sync-workerthat consumes daily HR employee batches fromhr-syncerand maintains a per-section ES index (spotlightorg-{siteID}-vN) keyed bysectId.stream.OrgSyncStream,subject.OrgSyncEmployeesUpsert) — Go names deliberately scoped to "org sync" to avoid clashing with the internal repo'sHRJOSbuilders; wire values still match hr-syncer's actual publish.DEV_MODEtoggle through the message + spotlight templates so single-node ES setups can run 1 shard / 0 replicas.Design
focin prod). Every fab site's search-sync-worker consumes from the sharedHR_{centralSiteID}stream. ThespotlightOrgCollectionstores both alocalSiteIDand anhrCentralSiteID: stream config + filter subjects target the central site (where hr-syncer publishes), while the durable consumer name is scoped by local site (spotlight-org-sync-{localSiteID}) so each fab site keeps its own cursor.{timestamp int64, employees: [...]}; each employee element is a full Employee struct from hr-syncer's internal repo. On our side we decode straight into[]SpotlightOrgIndex— JSON drops fields we don't declare, so we pick up only the nine org fields we care about.Nats-Encoding: zstdheader. Decompression is handled centrally inhandler.go'sAdd(a sharedcompression.gohelper with a pooledzstd.Decoderand a 16 MBWithDecoderMaxMemorycap). Collections never see the compressed frame; theCollection.BuildAction([]byte)interface stays codec-agnostic.sectId+ a renamedsectName) produce ES_updatebodies containing only the changed keys. Doc-merge preserves other stored fields without any painless script.SectID: within a batch, employees sharing the same section collapse to one action (last-wins). Empty-SectIDrows are skipped; all-empty batches return(nil, nil)so the handler acks with no ES write.HR_{centralSiteID}schema is owned byhr-syncer. search-sync-worker is a pure consumer and skips this stream in its bootstrap loop, matching how it already skipsINBOX(owned by inbox-worker).Commits
feat(stream,subject): HR wire infrastructure—stream.OrgSyncStream,subject.OrgSyncEmployeesUpsert.refactor(search-sync-worker): DEV_MODE toggle for message + spotlight templates— threaddevModethrough the two existing template builders (spotlight-orggets it too as part of commit 3).feat(search-sync-worker): spotlight-org collection consuming chat.hr.employees.upsert— newspotlightOrgCollection, sharedcompression.go, main.go wiring (SPOTLIGHT_ORG_INDEX+HR_CENTRAL_SITE_ID+DEV_MODEenv vars,HRbootstrap skip), integration test, docker-compose.New env vars for search-sync-worker
SPOTLIGHT_ORG_INDEX-vN)spotlightorg-foc-v1HR_CENTRAL_SITE_IDhr-syncerpublishes (e.g.foc)DEV_MODEfalse)Test plan
make test SERVICE=pkg/stream— passesmake test SERVICE=pkg/subject— passesmake test SERVICE=search-sync-worker— passes (spotlight-org unit tests + handler-level zstd decompression tests + all preexisting collection tests)make lint— cleango vet -tags=integration ./search-sync-worker/...— cleanmake test-integration SERVICE=search-sync-workeragainst the testcontainers harness — pending CI (TestSearchSyncSpotlightOrg_Integrationpublishes zstd-compressed batches withNats-Encoding: zstdheader and verifies doc upsert + doc-merge field preservation)Coordination with
hr-syncerWire-format contract Mat needs to match on the publish side:
chat.hr.{centralSiteID}.employees.upsertHR_{centralSiteID}{"timestamp": int64, "employees": [...]}where each employee has at minimum asectIdstring fieldNats-Encoding: zstdand publish the zstd-compressed body directly (no base64, no envelope wrapper).Spec / plan
docs/superpowers/specs/2026-05-11-spotlight-org-sync-design.mddocs/superpowers/plans/2026-05-11-spotlight-org-sync.mdNote: the spec and plan docs reflect an earlier iteration (envelope-based compression, single-site consumer) and haven't been updated for the central-site topology or header-based compression. The current implementation is authoritative.
https://claude.ai/code/session_01QLFefxiCHDP24LDLQzLLG2