diff --git a/cmd/genesis-writer/entities_track.go b/cmd/genesis-writer/entities_track.go index 2ab90525..4ea15927 100644 --- a/cmd/genesis-writer/entities_track.go +++ b/cmd/genesis-writer/entities_track.go @@ -94,7 +94,7 @@ type sourceTrack struct { IsUnlisted bool IsDownloadable bool IsOriginalAvailable bool - ReleaseDate *string + ReleaseDate *time.Time License *string ISRC *string ISWC *string @@ -157,7 +157,7 @@ func buildTrackMetadata(t sourceTrack, collaborators []int64) trackMetadataInner IsUnlisted: t.IsUnlisted, IsDownloadable: t.IsDownloadable, IsOriginalAvail: t.IsOriginalAvailable, - ReleaseDate: deref(t.ReleaseDate), + ReleaseDate: fmtReleaseDate(t.ReleaseDate), License: deref(t.License), ISRC: deref(t.ISRC), ISWC: deref(t.ISWC), @@ -211,6 +211,21 @@ func buildTrackMetadata(t sourceTrack, collaborators []int64) trackMetadataInner return inner } +// fmtReleaseDate emits RFC3339, the format the indexer's parseReleaseDate +// accepts. Selecting release_date::text instead yields Postgres's own +// "2026-09-06 22:06:00", which matches none of the accepted layouts, so the +// indexer silently fell back to block time. That is not a cosmetic date +// difference: a track whose release_date lands in the past is picked up by the +// scheduled-release publisher, which sets is_unlisted = false. On the +// 2026-08-07 snapshot 372 unlisted tracks with a future release date had the +// date rewritten and 368 of them were published early. +func fmtReleaseDate(t *time.Time) string { + if t == nil { + return "" + } + return t.UTC().Format(time.RFC3339) +} + func (w *Writer) writeTracks(ctx context.Context) error { // Pre-load collaborator lists so Track:Create metadata includes them, // which causes the ETL to create pending invites automatically. @@ -231,7 +246,7 @@ func (w *Writer) writeTracks(ctx context.Context) error { t.track_cid, t.cover_art, t.cover_art_sizes, t.preview_cid, t.is_unlisted, t.is_downloadable, t.is_original_available, - t.release_date::text, t.license, t.isrc, t.iswc, t.bpm, t.musical_key, + t.release_date, t.license, t.isrc, t.iswc, t.bpm, t.musical_key, t.is_custom_bpm, t.is_custom_musical_key, t.remix_of, t.stem_of, t.is_stream_gated, t.stream_conditions, diff --git a/cmd/genesis-writer/release_date_format_test.go b/cmd/genesis-writer/release_date_format_test.go new file mode 100644 index 00000000..1977fa10 --- /dev/null +++ b/cmd/genesis-writer/release_date_format_test.go @@ -0,0 +1,32 @@ +package main + +import ( + "testing" + "time" +) + +// The indexer's parseReleaseDate accepts RFC3339, RFC3339Nano and +// "Mon Jan 02 2006 15:04:05 GMT-0700" -- and nothing else. Selecting +// release_date::text yields Postgres's "2026-09-06 22:06:00", which matches +// none of them, so the indexer silently fell back to block time. A future +// release date rewritten into the past is then picked up by the +// scheduled-release publisher, which sets is_unlisted = false: on the +// 2026-08-07 snapshot that rewrote 372 dates and published 368 unlisted +// tracks early. +func TestReleaseDateIsEmittedInAnAcceptedLayout(t *testing.T) { + rd := time.Date(2026, 9, 6, 22, 6, 0, 0, time.UTC) + got := fmtReleaseDate(&rd) + + if _, err := time.Parse(time.RFC3339, got); err != nil { + t.Fatalf("fmtReleaseDate produced %q, which the indexer cannot parse: %v", got, err) + } + if got != "2026-09-06T22:06:00Z" { + t.Errorf("fmtReleaseDate = %q, want %q", got, "2026-09-06T22:06:00Z") + } + if got == "2026-09-06 22:06:00" { + t.Error("emitted Postgres text format, which no accepted layout matches") + } + if fmtReleaseDate(nil) != "" { + t.Errorf("nil release_date = %q, want empty so omitempty drops it", fmtReleaseDate(nil)) + } +} diff --git a/cmd/genesis-writer/step_order_test.go b/cmd/genesis-writer/step_order_test.go new file mode 100644 index 00000000..85c49e81 --- /dev/null +++ b/cmd/genesis-writer/step_order_test.go @@ -0,0 +1,46 @@ +package main + +import ( + "strings" + "testing" +) + +// The indexer validates references against the state a transaction lands on, +// so a step that emits rows pointing at entities a later step creates loses +// them outright -- silently, since row counts on the referencing table look +// plausible either way. +// +// Measured on the 2026-08-07 snapshot with comments running before events: +// all 69 Event comments were emitted by the writer and refused by the indexer +// with "event %d does not exist". +func TestStepOrderPutsReferencedEntitiesFirst(t *testing.T) { + order := (&Writer{cfg: &WriterConfig{}}).stepNames() + + idx := func(name string) int { + for i, n := range order { + if n == name { + return i + } + } + t.Fatalf("step %q not found in %v", name, order) + return -1 + } + + for _, dep := range []struct{ before, after, why string }{ + {"events", "comments", "a comment with entity_type=Event needs its event to exist"}, + {"events", "event subscriptions", "a subscription needs its target event to exist"}, + {"users", "tracks", "a track needs its owner"}, + {"tracks", "playlists", "playlist contents reference tracks"}, + {"comments", "comment reactions", "a reaction needs its comment"}, + {"comments", "comment pins", "a pin references a comment"}, + } { + if b, a := idx(dep.before), idx(dep.after); b > a { + t.Errorf("step %q runs after %q (positions %d, %d): %s", + dep.before, dep.after, b, a, dep.why) + } + } + + if strings.Join(order, ",") == "" { + t.Fatal("no steps registered") + } +} diff --git a/cmd/genesis-writer/writer.go b/cmd/genesis-writer/writer.go index a21dcb05..ed9fddbd 100644 --- a/cmd/genesis-writer/writer.go +++ b/cmd/genesis-writer/writer.go @@ -347,6 +347,93 @@ func (s *txCopySource) Err() error { return nil } // Run processes all entity types in dependency order, writes them as real // CometBFT blocks, then writes CometBFT state files. + +// writeStep is one phase of the run. Order matters: the indexer validates +// references against the state a transaction lands on, so a step emitting rows +// that point at entities a later step creates loses them silently. +type writeStep struct { + name string + skip bool + fn func(context.Context) error +} + +// steps is the ordered plan for a run. TestStepOrderPutsReferencedEntitiesFirst +// pins the dependencies that are not obvious from reading it. +func (w *Writer) steps() []writeStep { + return []writeStep{ + // Rewards first: it is the only step that depends on something outside + // this process (--core-dsn, pointed at the old chain, usually across a + // port-forward), and it is seconds of work against hours for the rest. + // Ordered last, an unreachable source failed the run only after all of + // that had completed. Nothing here depends on ordering — rewards + // reference only the pools this step creates, and no entity references + // a reward. + {"rewards", w.cfg.SkipRewards, w.writeRewards}, + + // Phase 1: Identity — users and their linked wallets + {"users", w.cfg.SkipUsers, w.writeUsers}, + {"associated wallets", w.cfg.SkipWallets, w.writeAssociatedWallets}, + {"dashboard wallet users", w.cfg.SkipWallets, w.writeDashboardWalletUsers}, + + // Phase 2: Content — tracks, collaborators, and playlists + {"tracks", w.cfg.SkipTracks, w.writeTracks}, + {"track collaborator approvals", w.cfg.SkipTracks, w.writeTrackCollaboratorApprovals}, + {"track downloads", w.cfg.SkipTracks, w.writeTrackDownloads}, + {"playlists", w.cfg.SkipPlaylists, w.writePlaylists}, + + // Phase 3: Social — relationships between users and content + {"follows", w.cfg.SkipSocial, w.writeFollows}, + {"saves", w.cfg.SkipSocial, w.writeSaves}, + {"reposts", w.cfg.SkipSocial, w.writeReposts}, + {"shares", w.cfg.SkipSocial, w.writeShares}, + {"subscriptions", w.cfg.SkipSocial, w.writeSubscriptions}, + {"muted users", w.cfg.SkipSocial, w.writeMutedUsers}, + + // Phase 4: Apps — developer apps and API grants + {"developer apps", w.cfg.SkipApps, w.writeDeveloperApps}, + {"grants", w.cfg.SkipApps, w.writeGrants}, + + // Phase 5: Events — before comments, which can hang off them. + // A comment carrying entity_type=Event is rejected outright if the + // event is not in place yet ("event %d does not exist"), and every + // comment transaction precedes every event transaction when this step + // runs later. On the 2026-08-07 snapshot that silently dropped all 69 + // Event comments -- the writer emitted them, the indexer refused them, + // and row counts on a 327k-row table hid it. + // Event subscriptions follow the events they point at for the same + // reason: the indexer rejects a subscription whose target event does + // not exist yet. + {"events", w.cfg.SkipEvents, w.writeEvents}, + {"event subscriptions", w.cfg.SkipEvents, w.writeEventSubscriptions}, + + // Phase 6: Comments — comments and reactions on content + {"comments", w.cfg.SkipComments, w.writeComments}, + {"comment reactions", w.cfg.SkipComments, w.writeCommentReactions}, + // A pin references both a comment and the track it is pinned to, so it + // has to follow the comments step and cannot run if either side was + // skipped. + {"comment pins", w.cfg.SkipComments || w.cfg.SkipTracks, w.writeCommentPins}, + + // Phase 7: Emails — encrypted emails and access grants + {"encrypted emails", w.cfg.SkipEmails, w.writeEncryptedEmails}, + {"email access", w.cfg.SkipEmails, w.writeEmailAccess}, + + // Phase 9: Activity — play count reconciliation and plays + {"play count reconciliation", w.cfg.SkipPlays, w.writePlayCountReconciliation}, + {"plays", w.cfg.SkipPlays, w.writePlays}, + } +} + +// stepNames returns the step order for tests. +func (w *Writer) stepNames() []string { + steps := w.steps() + names := make([]string, 0, len(steps)) + for _, s := range steps { + names = append(names, s.name) + } + return names +} + func (w *Writer) Run(ctx context.Context) error { start := time.Now() w.logger.Info("starting genesis write") @@ -436,65 +523,7 @@ func (w *Writer) Run(ctx context.Context) error { w.startBlockWriter(ctx) defer w.stopBlockWriter() //nolint:errcheck // explicit stop below captures the error - steps := []struct { - name string - skip bool - fn func(context.Context) error - }{ - // Rewards first: it is the only step that depends on something outside - // this process (--core-dsn, pointed at the old chain, usually across a - // port-forward), and it is seconds of work against hours for the rest. - // Ordered last, an unreachable source failed the run only after all of - // that had completed. Nothing here depends on ordering — rewards - // reference only the pools this step creates, and no entity references - // a reward. - {"rewards", w.cfg.SkipRewards, w.writeRewards}, - - // Phase 1: Identity — users and their linked wallets - {"users", w.cfg.SkipUsers, w.writeUsers}, - {"associated wallets", w.cfg.SkipWallets, w.writeAssociatedWallets}, - {"dashboard wallet users", w.cfg.SkipWallets, w.writeDashboardWalletUsers}, - - // Phase 2: Content — tracks, collaborators, and playlists - {"tracks", w.cfg.SkipTracks, w.writeTracks}, - {"track collaborator approvals", w.cfg.SkipTracks, w.writeTrackCollaboratorApprovals}, - {"track downloads", w.cfg.SkipTracks, w.writeTrackDownloads}, - {"playlists", w.cfg.SkipPlaylists, w.writePlaylists}, - - // Phase 3: Social — relationships between users and content - {"follows", w.cfg.SkipSocial, w.writeFollows}, - {"saves", w.cfg.SkipSocial, w.writeSaves}, - {"reposts", w.cfg.SkipSocial, w.writeReposts}, - {"shares", w.cfg.SkipSocial, w.writeShares}, - {"subscriptions", w.cfg.SkipSocial, w.writeSubscriptions}, - {"muted users", w.cfg.SkipSocial, w.writeMutedUsers}, - - // Phase 4: Apps — developer apps and API grants - {"developer apps", w.cfg.SkipApps, w.writeDeveloperApps}, - {"grants", w.cfg.SkipApps, w.writeGrants}, - - // Phase 5: Comments — comments and reactions on content - {"comments", w.cfg.SkipComments, w.writeComments}, - {"comment reactions", w.cfg.SkipComments, w.writeCommentReactions}, - // A pin references both a comment and the track it is pinned to, so it - // has to follow the comments step and cannot run if either side was - // skipped. - {"comment pins", w.cfg.SkipComments || w.cfg.SkipTracks, w.writeCommentPins}, - - // Phase 6: Emails — encrypted emails and access grants - {"encrypted emails", w.cfg.SkipEmails, w.writeEncryptedEmails}, - {"email access", w.cfg.SkipEmails, w.writeEmailAccess}, - - // Phase 7: Events - // Event subscriptions follow the events they point at: the indexer - // rejects a subscription whose target event does not exist yet. - {"events", w.cfg.SkipEvents, w.writeEvents}, - {"event subscriptions", w.cfg.SkipEvents, w.writeEventSubscriptions}, - - // Phase 9: Activity — play count reconciliation and plays - {"play count reconciliation", w.cfg.SkipPlays, w.writePlayCountReconciliation}, - {"plays", w.cfg.SkipPlays, w.writePlays}, - } + steps := w.steps() // Load completed steps for resume. completedSteps := make(map[string]bool)