From 8f4e08b6f167505c75e326923b2a49e7c296d01d Mon Sep 17 00:00:00 2001 From: Worph Date: Wed, 2 Sep 2026 00:07:02 +0200 Subject: [PATCH] rework domain field --- internal/api/admin_migrate.go | 26 +++ internal/api/server.go | 3 + internal/storage/domain_screen_migration.go | 193 ++++++++++++++++++ .../storage/domain_screen_migration_test.go | 140 +++++++++++++ 4 files changed, 362 insertions(+) create mode 100644 internal/storage/domain_screen_migration.go create mode 100644 internal/storage/domain_screen_migration_test.go diff --git a/internal/api/admin_migrate.go b/internal/api/admin_migrate.go index e27f127..65bb211 100644 --- a/internal/api/admin_migrate.go +++ b/internal/api/admin_migrate.go @@ -28,3 +28,29 @@ func (s *Server) handleMigrateDualRoots(w http.ResponseWriter, r *http.Request) "fixed": fixed, }) } + +// handleMigrateDomainScreen triggers the one-shot sweep that rewrites the +// retired `domain=film|tv` values to `screen` and backfills `workForm` from +// `contentKind` (METADATA_KEYS.md §14.17). See +// storage.MigrateDomainScreen for why the vocabulary change is paid in the +// data rather than with a read-side alias. +// +// Returns {fixed: N} — records changed, not records examined. Safe to call +// repeatedly: a record already on the current vocabulary is skipped. +func (s *Server) handleMigrateDomainScreen(w http.ResponseWriter, r *http.Request) { + if !s.storage.IsConnected() { + writeError(w, http.StatusServiceUnavailable, "storage not connected") + return + } + + fixed, err := s.storage.MigrateDomainScreenWithTimeout() + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + + writeJSON(w, http.StatusOK, map[string]interface{}{ + "success": true, + "fixed": fixed, + }) +} diff --git a/internal/api/server.go b/internal/api/server.go index fa7a351..7d82e17 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -253,6 +253,9 @@ func (s *Server) setupRoutes() { // (the dual-root pattern caused by historical writes that bypassed CID // resolution). s.router.HandleFunc("/api/admin/migrate-dual-roots", s.handleMigrateDualRoots).Methods("POST") + // One-shot vocabulary sweep: domain film|tv → screen, workForm backfill + // (METADATA_KEYS.md §14.17). Idempotent. + s.router.HandleFunc("/api/admin/migrate-domain-screen", s.handleMigrateDomainScreen).Methods("POST") // SSE event streams. Mediated mirror of the Redis Streams; consumers // reach them over HTTP instead of touching Redis directly. Auth-bypass diff --git a/internal/storage/domain_screen_migration.go b/internal/storage/domain_screen_migration.go new file mode 100644 index 0000000..bfb619b --- /dev/null +++ b/internal/storage/domain_screen_migration.go @@ -0,0 +1,193 @@ +package storage + +import ( + "context" + "fmt" + "time" +) + +// The one-shot corpus sweep for METADATA_KEYS.md §14.17: the `film` + `tv` +// domains merged into `screen`, and the standalone-vs-serial split they were +// really carrying moved to the new `workForm` key. +// +// # Why a sweep and not a read-side alias +// +// `domain` is **stored, not derived on read** — that is what lets the +// `contentKind` vocabulary stay open (a peer that has never heard of a kind +// still routes the record correctly). The price of storing it is that a +// vocabulary change has to be paid once, in the data: every consumer's filter +// is an exact match, so a record left saying `film` is invisible to a +// `domain:screen` wall, and teaching every consumer to accept both spellings +// forever is exactly the alternation the merge exists to delete. +// +// So the values are rewritten at rest, here, and no alias vocabulary survives +// in query filters. (meta-search additionally normalises on the way into its +// index, because a record can still arrive from a peer that has not been swept +// — that is a courtesy for un-swept peers, not a second source of truth.) +// +// # Contract +// +// Idempotent: a record already carrying `screen` and a `workForm` is skipped, +// so the endpoint is safe to call repeatedly and safe to call on a fresh box. +// Returns the number of records **changed**, not the number examined. + +// legacyDomainReplacements maps a retired `domain` value to its replacement. +var legacyDomainReplacements = map[string]string{ + "film": "screen", + "tv": "screen", +} + +// workFormByContentKind is the `contentKind` → `workForm` table from +// METADATA_KEYS.md §1. +// +// ⚠ Mirror of the Rust table in `meta-feeder-sdk::domain` (and of +// meta-search's copy). It exists here only to backfill the pre-merge corpus — +// meta-core never classifies a record it is given, and nothing else in Go reads +// it. If the tables ever disagree, the writers win: this one is a one-shot. +var workFormByContentKind = map[string]string{ + // screen + "movie": "standalone", + "series": "serial", + "episode": "serial", + // music — an album is a closed work, not an ongoing run. + "track": "standalone", + "album": "standalone", + "artist": "standalone", + "musicVideo": "standalone", + "djMix": "standalone", + "liveSet": "standalone", + // literature + "book": "standalone", + "audiobook": "standalone", + "comic": "serial", + "manga": "serial", + "magazine": "serial", + // science + "paper": "standalone", + // ⚠ `pack` is deliberately absent: a season pack is `serial`, an album + // release `standalone`, and only the writer knows which. A pack that + // reaches this sweep without a `workForm` keeps none — inventing one would + // be a guess written permanently into the record. +} + +// domainScreenPlan is one record's pending rewrite. Empty strings mean "leave +// this field alone". +type domainScreenPlan struct { + hashID string + domain string + workForm string +} + +// MigrateDomainScreen rewrites `domain=film|tv` to `domain=screen` and +// backfills `workForm` from `contentKind` across the whole corpus. +// +// Reads in two batched MGET passes (never a per-record SCAN — a single record +// read costs a full keyspace sweep on this store), then writes only the records +// that actually change. +func (c *Client) MigrateDomainScreen(ctx context.Context) (int, error) { + plans, err := c.planDomainScreen(ctx) + if err != nil { + return 0, err + } + // Writes take the client lock themselves, so they happen after the read + // phase has released it. + fixed := 0 + for _, p := range plans { + if p.domain != "" { + if err := c.SetProperty(p.hashID, "domain", p.domain); err != nil { + return fixed, fmt.Errorf("set domain on %s: %w", p.hashID, err) + } + } + if p.workForm != "" { + if err := c.SetProperty(p.hashID, "workForm", p.workForm); err != nil { + return fixed, fmt.Errorf("set workForm on %s: %w", p.hashID, err) + } + } + fixed++ + } + return fixed, nil +} + +// planDomainScreen is the read half: what would change, without changing it. +func (c *Client) planDomainScreen(ctx context.Context) ([]domainScreenPlan, error) { + c.mu.RLock() + defer c.mu.RUnlock() + + if c.client == nil { + return nil, fmt.Errorf("not connected") + } + + hashIDs, err := c.client.SMembers(ctx, c.buildIndexKey()).Result() + if err != nil { + return nil, fmt.Errorf("smembers index: %w", err) + } + if len(hashIDs) == 0 { + return nil, nil + } + + // Three keys per record, in a fixed order so the MGET response decodes by + // position: keys[i*3+n] belongs to hashIDs[i]. + const fieldsPerFile = 3 + fields := [fieldsPerFile]string{"domain", "workForm", "contentKind"} + keys := make([]string, 0, len(hashIDs)*fieldsPerFile) + for _, h := range hashIDs { + prefix := c.buildKeyPrefix(h) + for _, f := range fields { + keys = append(keys, prefix+f) + } + } + + // Chunked so one MGET never carries the whole corpus. + const chunk = 3000 + values := make([]interface{}, 0, len(keys)) + for start := 0; start < len(keys); start += chunk { + end := start + chunk + if end > len(keys) { + end = len(keys) + } + got, err := c.client.MGet(ctx, keys[start:end]...).Result() + if err != nil { + return nil, fmt.Errorf("mget: %w", err) + } + values = append(values, got...) + } + + str := func(i int) string { + if i >= len(values) { + return "" + } + s, _ := values[i].(string) + return s + } + + plans := make([]domainScreenPlan, 0) + for i, h := range hashIDs { + base := i * fieldsPerFile + domain, workForm, contentKind := str(base), str(base+1), str(base+2) + + plan := domainScreenPlan{hashID: h} + if replacement, ok := legacyDomainReplacements[domain]; ok { + plan.domain = replacement + } + // A stored `workForm` is authoritative — a `pack` writer knew something + // the table cannot. Only an absent one is filled in. + if workForm == "" { + if form, ok := workFormByContentKind[contentKind]; ok { + plan.workForm = form + } + } + if plan.domain != "" || plan.workForm != "" { + plans = append(plans, plan) + } + } + return plans, nil +} + +// MigrateDomainScreenWithTimeout is the convenience wrapper used by the API +// handler. Generous timeout: the read phase is two batched passes, but the +// write phase is one round-trip per changed field. +func (c *Client) MigrateDomainScreenWithTimeout() (int, error) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute) + defer cancel() + return c.MigrateDomainScreen(ctx) +} diff --git a/internal/storage/domain_screen_migration_test.go b/internal/storage/domain_screen_migration_test.go new file mode 100644 index 0000000..792ca1d --- /dev/null +++ b/internal/storage/domain_screen_migration_test.go @@ -0,0 +1,140 @@ +package storage + +import ( + "context" + "testing" +) + +// seedRecord writes one record's classification fields straight through +// SetProperty, which is what a real writer uses (it maintains the field index +// and the root index the sweep reads from). +func seedRecord(t *testing.T, c *Client, hashID string, fields map[string]string) { + t.Helper() + for k, v := range fields { + if err := c.SetProperty(hashID, k, v); err != nil { + t.Fatalf("seed %s/%s: %v", hashID, k, err) + } + } +} + +// The merge (METADATA_KEYS.md §14.17) is paid in the data: `domain` is stored, +// not derived on read, so a record left saying `film` is invisible to a +// `domain:screen` wall. This is the sweep that pays it. +func TestMigrateDomainScreen_RewritesLegacyDomainsAndBackfillsWorkForm(t *testing.T) { + c, _ := newTestClient(t, "") + + seedRecord(t, c, "rec-movie", map[string]string{ + "domain": "film", + "contentKind": "movie", + }) + seedRecord(t, c, "rec-episode", map[string]string{ + "domain": "tv", + "contentKind": "episode", + }) + // Already current, and already carries both keys: must be left alone. + seedRecord(t, c, "rec-current", map[string]string{ + "domain": "screen", + "contentKind": "series", + "workForm": "serial", + }) + // A different domain entirely — the sweep must not touch it, but it is + // still owed the `workForm` backfill. + seedRecord(t, c, "rec-book", map[string]string{ + "domain": "literature", + "contentKind": "book", + }) + + fixed, err := c.MigrateDomainScreen(context.Background()) + if err != nil { + t.Fatalf("migrate: %v", err) + } + if fixed != 3 { + t.Errorf("fixed = %d, want 3 (the two legacy rows + the book's backfill)", fixed) + } + + for _, tc := range []struct { + hashID string + domain string + workForm string + }{ + {"rec-movie", "screen", "standalone"}, + {"rec-episode", "screen", "serial"}, + {"rec-current", "screen", "serial"}, + {"rec-book", "literature", "standalone"}, + } { + m, err := c.GetMetadataFlat(tc.hashID) + if err != nil { + t.Fatalf("read %s: %v", tc.hashID, err) + } + if m["domain"] != tc.domain { + t.Errorf("%s domain = %q, want %q", tc.hashID, m["domain"], tc.domain) + } + if m["workForm"] != tc.workForm { + t.Errorf("%s workForm = %q, want %q", tc.hashID, m["workForm"], tc.workForm) + } + } +} + +// A second pass must find nothing to do — the endpoint is called by hand on +// each box, and "run it twice" must not mean "write the whole corpus twice". +func TestMigrateDomainScreen_IsIdempotent(t *testing.T) { + c, _ := newTestClient(t, "mm:") + + seedRecord(t, c, "rec-1", map[string]string{ + "domain": "tv", + "contentKind": "episode", + }) + + if _, err := c.MigrateDomainScreen(context.Background()); err != nil { + t.Fatalf("first pass: %v", err) + } + fixed, err := c.MigrateDomainScreen(context.Background()) + if err != nil { + t.Fatalf("second pass: %v", err) + } + if fixed != 0 { + t.Errorf("second pass fixed = %d, want 0", fixed) + } +} + +// `pack` is the one kind neither routing table can decide — a season pack is +// `serial`, an album release `standalone`. A pack that reaches the sweep +// without a `workForm` must keep none: inventing one writes a guess into the +// record permanently. +func TestMigrateDomainScreen_NeverGuessesAPacksWorkForm(t *testing.T) { + c, _ := newTestClient(t, "") + + seedRecord(t, c, "rec-pack", map[string]string{ + "domain": "tv", + "contentKind": "pack", + }) + // A writer that DID know keeps what it wrote. + seedRecord(t, c, "rec-album", map[string]string{ + "domain": "music", + "contentKind": "pack", + "workForm": "standalone", + }) + + if _, err := c.MigrateDomainScreen(context.Background()); err != nil { + t.Fatalf("migrate: %v", err) + } + + m, err := c.GetMetadataFlat("rec-pack") + if err != nil { + t.Fatalf("read: %v", err) + } + if m["domain"] != "screen" { + t.Errorf("domain = %q, want screen (the domain WAS knowable)", m["domain"]) + } + if _, ok := m["workForm"]; ok { + t.Errorf("workForm = %q, want absent", m["workForm"]) + } + + a, err := c.GetMetadataFlat("rec-album") + if err != nil { + t.Fatalf("read: %v", err) + } + if a["workForm"] != "standalone" { + t.Errorf("workForm = %q, want standalone (writer-supplied, never overwritten)", a["workForm"]) + } +}