diff --git a/cmd/genesis-writer/entities_comment.go b/cmd/genesis-writer/entities_comment.go index 71fc6005..5fbee98c 100644 --- a/cmd/genesis-writer/entities_comment.go +++ b/cmd/genesis-writer/entities_comment.go @@ -20,6 +20,13 @@ type commentMetadata struct { TrackTimestampS *int `json:"track_timestamp_s,omitempty"` Mentions []int64 `json:"mentions,omitempty"` CreatedAt string `json:"created_at,omitempty"` + // Fan-club text post fields. The indexer reads both with a default that + // matches an unset column (false / NULL), so `omitempty` cannot flip a value + // on read and only the 39 members-only and 15 video comments carry them. + // is_members_only is emitted verbatim from the source: every row that sets + // it is entity_type='FanClub', which is what validateCommentWrite requires. + IsMembersOnly bool `json:"is_members_only,omitempty"` + VideoURL string `json:"video_url,omitempty"` // Always serialized: `omitempty` would drop a false value and the indexer // cannot tell "absent" from "not deleted". IsDelete bool `json:"is_delete"` @@ -35,6 +42,8 @@ type sourceComment struct { TrackTimestampS *int CreatedAt time.Time IsDelete bool + IsMembersOnly bool + VideoURL *string } // writeComments emits root comments and replies in two passes. @@ -106,7 +115,7 @@ func (w *Writer) writeCommentPass( `SELECT count(*) FROM comments c JOIN users u ON u.user_id = c.user_id AND u.is_current = true AND u.wallet IS NOT NULL AND u.wallet <> '' `+where, - `SELECT c.comment_id, c.text, c.user_id, COALESCE(LOWER(u.wallet), ''), c.entity_id, c.entity_type, c.track_timestamp_s, c.created_at, c.is_delete + `SELECT c.comment_id, c.text, c.user_id, COALESCE(LOWER(u.wallet), ''), c.entity_id, c.entity_type, c.track_timestamp_s, c.created_at, c.is_delete, c.is_members_only, c.video_url FROM comments c JOIN users u ON u.user_id = c.user_id AND u.is_current = true AND u.wallet IS NOT NULL AND u.wallet <> '' `+where+` @@ -115,7 +124,7 @@ func (w *Writer) writeCommentPass( ORDER BY c.created_at, c.comment_id`, func(rows pgx.Rows) (sourceComment, error) { var c sourceComment - err := rows.Scan(&c.CommentID, &c.Text, &c.UserID, &c.UserWallet, &c.EntityID, &c.EntityType, &c.TrackTimestampS, &c.CreatedAt, &c.IsDelete) + err := rows.Scan(&c.CommentID, &c.Text, &c.UserID, &c.UserWallet, &c.EntityID, &c.EntityType, &c.TrackTimestampS, &c.CreatedAt, &c.IsDelete, &c.IsMembersOnly, &c.VideoURL) return c, err }, func(ctx context.Context, c sourceComment) error { @@ -126,6 +135,8 @@ func (w *Writer) writeCommentPass( EntityType: c.EntityType, TrackTimestampS: c.TrackTimestampS, CreatedAt: c.CreatedAt.UTC().Format(time.RFC3339), + IsMembersOnly: c.IsMembersOnly, + VideoURL: deref(c.VideoURL), } // Attach parent comment if this is a reply. diff --git a/cmd/genesis-writer/entities_playlist.go b/cmd/genesis-writer/entities_playlist.go index b8a7b066..b70323b3 100644 --- a/cmd/genesis-writer/entities_playlist.go +++ b/cmd/genesis-writer/entities_playlist.go @@ -22,9 +22,11 @@ type playlistMetadataInner struct { Description string `json:"description,omitempty"` IsAlbum bool `json:"is_album,omitempty"` IsPrivate bool `json:"is_private,omitempty"` + IsImageAutogenerated bool `json:"is_image_autogenerated,omitempty"` PlaylistImageSizesHash string `json:"playlist_image_sizes_multihash,omitempty"` PlaylistContents interface{} `json:"playlist_contents,omitempty"` ReleaseDate string `json:"release_date,omitempty"` + IsScheduledRelease bool `json:"is_scheduled_release,omitempty"` IsStreamGated bool `json:"is_stream_gated,omitempty"` StreamConditions interface{} `json:"stream_conditions,omitempty"` UPC string `json:"upc,omitempty"` @@ -93,32 +95,34 @@ func preloadRemovedPlaylistTracks(ctx context.Context, db *pgxpool.Pool) (map[in } type sourcePlaylist struct { - PlaylistID int64 - PlaylistOwnerID int64 - OwnerWallet string - PlaylistName *string - Description *string - IsAlbum bool - IsPrivate bool - MetadataMultihash *string - ImageSizesMultihash *string - ImageMultihash *string - PlaylistContents []byte // JSONB - UPC *string - DDEXApp *string - ParentalWarningType *string - ReleaseDate *string - IsStreamGated bool - StreamConditions []byte // JSONB - DDEXReleaseIDs []byte // JSONB - Artists []byte // JSONB - CopyrightLine []byte // JSONB - ProducerCopyright []byte // JSONB - RouteSlug *string - RouteTitleSlug *string - RouteCollisionID *int - IsDelete bool - CreatedAt time.Time + PlaylistID int64 + PlaylistOwnerID int64 + OwnerWallet string + PlaylistName *string + Description *string + IsAlbum bool + IsPrivate bool + IsImageAutogenerated bool + MetadataMultihash *string + ImageSizesMultihash *string + ImageMultihash *string + PlaylistContents []byte // JSONB + UPC *string + DDEXApp *string + ParentalWarningType *string + ReleaseDate *string + IsScheduledRelease bool + IsStreamGated bool + StreamConditions []byte // JSONB + DDEXReleaseIDs []byte // JSONB + Artists []byte // JSONB + CopyrightLine []byte // JSONB + ProducerCopyright []byte // JSONB + RouteSlug *string + RouteTitleSlug *string + RouteCollisionID *int + IsDelete bool + CreatedAt time.Time } func (w *Writer) writePlaylists(ctx context.Context) error { @@ -139,10 +143,10 @@ func (w *Writer) writePlaylists(ctx context.Context) error { `SELECT p.playlist_id, p.playlist_owner_id, COALESCE(LOWER(u.wallet), ''), p.playlist_name, p.description, - p.is_album, p.is_private, + p.is_album, p.is_private, p.is_image_autogenerated, p.metadata_multihash, p.playlist_image_sizes_multihash, p.playlist_image_multihash, p.playlist_contents, p.upc, p.ddex_app, p.parental_warning_type, - p.release_date::text, p.is_stream_gated, p.stream_conditions, + p.release_date::text, p.is_scheduled_release, p.is_stream_gated, p.stream_conditions, p.ddex_release_ids, p.artists, p.copyright_line, p.producer_copyright_line, r.slug, r.title_slug, r.collision_id, p.is_delete, @@ -163,10 +167,10 @@ func (w *Writer) writePlaylists(ctx context.Context) error { err := rows.Scan( &p.PlaylistID, &p.PlaylistOwnerID, &p.OwnerWallet, &p.PlaylistName, &p.Description, - &p.IsAlbum, &p.IsPrivate, + &p.IsAlbum, &p.IsPrivate, &p.IsImageAutogenerated, &p.MetadataMultihash, &p.ImageSizesMultihash, &p.ImageMultihash, &p.PlaylistContents, &p.UPC, &p.DDEXApp, &p.ParentalWarningType, - &p.ReleaseDate, &p.IsStreamGated, &p.StreamConditions, + &p.ReleaseDate, &p.IsScheduledRelease, &p.IsStreamGated, &p.StreamConditions, &p.DDEXReleaseIDs, &p.Artists, &p.CopyrightLine, &p.ProducerCopyright, &p.RouteSlug, &p.RouteTitleSlug, &p.RouteCollisionID, &p.IsDelete, @@ -181,12 +185,14 @@ func (w *Writer) writePlaylists(ctx context.Context) error { Description: deref(p.Description), IsAlbum: p.IsAlbum, IsPrivate: p.IsPrivate, + IsImageAutogenerated: p.IsImageAutogenerated, PlaylistImageSizesHash: deref(p.ImageSizesMultihash), PlaylistImageHash: deref(p.ImageMultihash), UPC: deref(p.UPC), DDEXApp: deref(p.DDEXApp), ParentalWarningType: deref(p.ParentalWarningType), ReleaseDate: deref(p.ReleaseDate), + IsScheduledRelease: p.IsScheduledRelease, IsStreamGated: p.IsStreamGated, RouteSlug: deref(p.RouteSlug), RouteTitleSlug: deref(p.RouteTitleSlug), diff --git a/cmd/genesis-writer/entities_social.go b/cmd/genesis-writer/entities_social.go index 624267f3..8b6dce8d 100644 --- a/cmd/genesis-writer/entities_social.go +++ b/cmd/genesis-writer/entities_social.go @@ -25,6 +25,24 @@ type socialMeta struct { IsDelete bool `json:"is_delete"` } +// saveMeta is socialMeta plus is_save_of_repost, which records that the save was +// made from a repost in someone's feed rather than from the item itself. The +// indexer reads it in insertSave with a default of false, so `omitempty` cannot +// flip a value on read and the key rides only on the 28,155 source rows that +// actually set it rather than on every save. +type saveMeta struct { + socialMeta + IsSaveOfRepost bool `json:"is_save_of_repost,omitempty"` +} + +// repostMeta is the repost counterpart of saveMeta: is_repost_of_repost marks a +// repost made from another repost. Same default-false read in insertRepost, same +// reasoning for `omitempty`; 21,366 source rows set it. +type repostMeta struct { + socialMeta + IsRepostOfRepost bool `json:"is_repost_of_repost,omitempty"` +} + func fmtCreatedAt(t time.Time) string { return t.UTC().Format(time.RFC3339) } @@ -79,23 +97,27 @@ func (w *Writer) writeSaves(ctx context.Context) error { saveType string createdAt time.Time isDelete bool + isSaveOfRepost bool } return processBatched(ctx, w, "saves", `SELECT count(*) FROM saves s JOIN users u ON u.user_id = s.user_id AND u.is_current = true AND u.wallet IS NOT NULL AND u.wallet <> '' WHERE s.is_current = true`, - `SELECT s.user_id, s.save_item_id, LOWER(u.wallet), s.save_type, s.created_at, s.is_delete + `SELECT s.user_id, s.save_item_id, LOWER(u.wallet), s.save_type, s.created_at, s.is_delete, s.is_save_of_repost FROM saves s JOIN users u ON u.user_id = s.user_id AND u.is_current = true AND u.wallet IS NOT NULL AND u.wallet <> '' WHERE s.is_current = true ORDER BY s.user_id, s.save_item_id`, func(rows pgx.Rows) (save, error) { var s save - err := rows.Scan(&s.userID, &s.itemID, &s.wallet, &s.saveType, &s.createdAt, &s.isDelete) + err := rows.Scan(&s.userID, &s.itemID, &s.wallet, &s.saveType, &s.createdAt, &s.isDelete, &s.isSaveOfRepost) return s, err }, func(ctx context.Context, s save) error { - metaJSON, err := json.Marshal(socialMeta{CreatedAt: fmtCreatedAt(s.createdAt), IsDelete: s.isDelete}) + metaJSON, err := json.Marshal(saveMeta{ + socialMeta: socialMeta{CreatedAt: fmtCreatedAt(s.createdAt), IsDelete: s.isDelete}, + IsSaveOfRepost: s.isSaveOfRepost, + }) if err != nil { return fmt.Errorf("marshal save metadata: %w", err) } @@ -114,28 +136,32 @@ func (w *Writer) writeSaves(ctx context.Context) error { func (w *Writer) writeReposts(ctx context.Context) error { type repost struct { - userID, itemID int64 - wallet string - repostType string - createdAt time.Time - isDelete bool + userID, itemID int64 + wallet string + repostType string + createdAt time.Time + isDelete bool + isRepostOfRepost bool } return processBatched(ctx, w, "reposts", `SELECT count(*) FROM reposts r JOIN users u ON u.user_id = r.user_id AND u.is_current = true AND u.wallet IS NOT NULL AND u.wallet <> '' WHERE r.is_current = true`, - `SELECT r.user_id, r.repost_item_id, COALESCE(LOWER(u.wallet), ''), r.repost_type, r.created_at, r.is_delete + `SELECT r.user_id, r.repost_item_id, COALESCE(LOWER(u.wallet), ''), r.repost_type, r.created_at, r.is_delete, r.is_repost_of_repost FROM reposts r JOIN users u ON u.user_id = r.user_id AND u.is_current = true AND u.wallet IS NOT NULL AND u.wallet <> '' WHERE r.is_current = true ORDER BY r.user_id, r.repost_item_id`, func(rows pgx.Rows) (repost, error) { var r repost - err := rows.Scan(&r.userID, &r.itemID, &r.wallet, &r.repostType, &r.createdAt, &r.isDelete) + err := rows.Scan(&r.userID, &r.itemID, &r.wallet, &r.repostType, &r.createdAt, &r.isDelete, &r.isRepostOfRepost) return r, err }, func(ctx context.Context, r repost) error { - metaJSON, err := json.Marshal(socialMeta{CreatedAt: fmtCreatedAt(r.createdAt), IsDelete: r.isDelete}) + metaJSON, err := json.Marshal(repostMeta{ + socialMeta: socialMeta{CreatedAt: fmtCreatedAt(r.createdAt), IsDelete: r.isDelete}, + IsRepostOfRepost: r.isRepostOfRepost, + }) if err != nil { return fmt.Errorf("marshal repost metadata: %w", err) } diff --git a/cmd/genesis-writer/entities_state_fields_test.go b/cmd/genesis-writer/entities_state_fields_test.go new file mode 100644 index 00000000..de37a622 --- /dev/null +++ b/cmd/genesis-writer/entities_state_fields_test.go @@ -0,0 +1,309 @@ +package main + +import ( + "context" + "os" + "testing" + "time" + + corev1 "github.com/OpenAudio/go-openaudio/pkg/api/core/v1" + etldb "github.com/OpenAudio/go-openaudio/pkg/etl/db" + em "github.com/OpenAudio/go-openaudio/pkg/etl/processors/entity_manager" + "github.com/jackc/pgx/v5/pgxpool" + "go.uber.org/zap" +) + +// stateFieldsSrcSchema holds the Discovery-Provider-shaped source tables for +// this test, for the same reasons as srcSchema in entities_comment_pin_test.go: +// a schema needs no CREATE DATABASE right and keeps the source out of public, +// where the ETL migrations run. +const stateFieldsSrcSchema = "genesis_writer_state_fields_src_test" + +// TestWriterEmitsStateFieldsTheIndexerReads covers a bug class that row-count +// parity is blind to: the indexer reads a metadata key the writer never emits, +// so the column silently lands on its default and the counts still match. +// +// Each subtest replays what the writer emits through the real migration +// dispatcher — the same handler set indexer.go builds — and asserts the column, +// because asserting the emitted JSON alone would not catch a metadata key the +// indexer spells differently. +// +// Source row counts on a production clone (audius_discovery_2026_08_07), i.e. +// what each field is worth: +// +// saves.is_save_of_repost 28,155 +// reposts.is_repost_of_repost 21,366 +// playlists.is_image_autogenerated 4,865 +// playlists.is_scheduled_release 566 +// comments.is_members_only 39 +// comments.video_url 15 +func TestWriterEmitsStateFieldsTheIndexerReads(t *testing.T) { + dbURL := os.Getenv("ETL_TEST_DB_URL") + if dbURL == "" { + t.Skip("ETL_TEST_DB_URL not set, skipping database test") + } + ctx := context.Background() + logger := zap.NewNop() + + const ( + ownerID = int64(9101) + trackID = int64(9201) + playlistA = int64(9301) // autogenerated image + scheduled release + playlistB = int64(9302) // neither, the negative control + commentA = int64(9401) // FanClub: members-only + video + commentB = int64(9402) // plain track comment, the negative control + fanClubID = int64(9501) + videoURL = "https://example.com/clip.mp4" + blockHash = "state-fields-block" + // Digits only: the writer lowercases the source wallet, so a wallet with + // no letters cannot fail ValidateSigner for reasons unrelated to this test. + ownerWallet = "0x9111111111111111111111111111111111111111" + ) + createdAt := time.Date(2025, 6, 1, 12, 0, 0, 0, time.UTC) + + if err := etldb.RunMigrations(logger, dbURL, true); err != nil { + t.Fatalf("run etl migrations: %v", err) + } + dst, err := pgxpool.New(ctx, dbURL) + if err != nil { + t.Fatalf("connect etl db: %v", err) + } + defer dst.Close() + + exec := func(pool *pgxpool.Pool, sql string, args ...any) { + t.Helper() + if _, err := pool.Exec(ctx, sql, args...); err != nil { + t.Fatalf("exec %q: %v", sql, err) + } + } + + // ---- indexed state the earlier migration steps would already have written -- + exec(dst, `INSERT INTO blocks (blockhash, parenthash, number) VALUES ($1, '', 1) + ON CONFLICT (blockhash) DO NOTHING`, blockHash) + exec(dst, `INSERT INTO users (user_id, handle, handle_lc, wallet, is_current, is_verified, is_deactivated, is_available, created_at, updated_at, txhash) + VALUES ($1, 'owner', 'owner', $2, true, false, false, true, now(), now(), '')`, ownerID, ownerWallet) + exec(dst, `INSERT INTO tracks (track_id, owner_id, title, is_current, is_delete, track_segments, created_at, updated_at, txhash) + VALUES ($1, $2, 'Saved', true, false, '[]', now(), now(), '')`, trackID, ownerID) + + // ---- source snapshot ------------------------------------------------------- + exec(dst, `DROP SCHEMA IF EXISTS `+stateFieldsSrcSchema+` CASCADE`) + exec(dst, `CREATE SCHEMA `+stateFieldsSrcSchema) + // Deferred rather than t.Cleanup so it runs before dst.Close above. + defer func() { + if _, err := dst.Exec(context.Background(), `DROP SCHEMA IF EXISTS `+stateFieldsSrcSchema+` CASCADE`); err != nil { + t.Logf("drop source schema: %v", err) + } + }() + + srcCfg, err := pgxpool.ParseConfig(dbURL) + if err != nil { + t.Fatalf("parse source dsn: %v", err) + } + srcCfg.ConnConfig.RuntimeParams["search_path"] = stateFieldsSrcSchema + src, err := pgxpool.NewWithConfig(ctx, srcCfg) + if err != nil { + t.Fatalf("connect source schema: %v", err) + } + defer src.Close() + + exec(src, `CREATE TABLE users (user_id bigint, wallet text, is_current boolean)`) + exec(src, `INSERT INTO users VALUES ($1, $2, true)`, ownerID, ownerWallet) + + // The migration dispatcher, built the way indexer.go builds it: production + // handlers first, then the genesis overrides. + dispatcher := em.NewDispatcher(logger) + dispatcher.Register(em.Save()) + dispatcher.Register(em.Repost()) + dispatcher.Register(em.PlaylistCreate()) + dispatcher.Register(em.CommentCreate()) + em.RegisterMigrationOverrides(dispatcher) + + replay := func(t *testing.T, tx *corev1.ManageEntityLegacyMigration) { + t.Helper() + params := em.NewParams(&corev1.ManageEntityLegacy{ + UserId: tx.GetUserId(), + EntityType: tx.GetEntityType(), + EntityId: tx.GetEntityId(), + Action: tx.GetAction(), + Metadata: tx.GetMetadata(), + Signature: tx.GetSignature(), + Signer: tx.GetSigner(), + Nonce: tx.GetNonce(), + }, 1, createdAt, blockHash, "txhash", dst, logger) + if err := dispatcher.Dispatch(ctx, params); err != nil { + t.Fatalf("indexing %s/%s %d: %v", tx.GetEntityType(), tx.GetAction(), tx.GetEntityId(), err) + } + } + + // --- saves.is_save_of_repost --------------------------------------------- + t.Run("save_is_save_of_repost", func(t *testing.T) { + exec(src, `CREATE TABLE saves (user_id bigint, save_item_id bigint, save_type text, + created_at timestamp, is_delete boolean, is_save_of_repost boolean, is_current boolean)`) + exec(src, `INSERT INTO saves VALUES ($1, $2, 'track', $3, false, true, true)`, ownerID, trackID, createdAt) + + w := newTestWriter(t, src) + if err := w.writeSaves(ctx); err != nil { + t.Fatalf("writeSaves: %v", err) + } + txs := decodeMigrationTxs(t, w.blockTxs, "Track", "Save") + if len(txs) != 1 { + t.Fatalf("emitted %d Save transactions, want 1", len(txs)) + } + replay(t, txs[0]) + + var got bool + if err := dst.QueryRow(ctx, + `SELECT is_save_of_repost FROM saves WHERE user_id = $1 AND save_item_id = $2 AND is_current = true`, + ownerID, trackID).Scan(&got); err != nil { + t.Fatalf("query save: %v", err) + } + if !got { + t.Errorf("saves.is_save_of_repost = false, want true (the source row sets it; "+ + "metadata was %s)", txs[0].GetMetadata()) + } + }) + + // --- reposts.is_repost_of_repost ------------------------------------------ + t.Run("repost_is_repost_of_repost", func(t *testing.T) { + exec(src, `CREATE TABLE reposts (user_id bigint, repost_item_id bigint, repost_type text, + created_at timestamp, is_delete boolean, is_repost_of_repost boolean, is_current boolean)`) + exec(src, `INSERT INTO reposts VALUES ($1, $2, 'track', $3, false, true, true)`, ownerID, trackID, createdAt) + + w := newTestWriter(t, src) + if err := w.writeReposts(ctx); err != nil { + t.Fatalf("writeReposts: %v", err) + } + txs := decodeMigrationTxs(t, w.blockTxs, "Track", "Repost") + if len(txs) != 1 { + t.Fatalf("emitted %d Repost transactions, want 1", len(txs)) + } + replay(t, txs[0]) + + var got bool + if err := dst.QueryRow(ctx, + `SELECT is_repost_of_repost FROM reposts WHERE user_id = $1 AND repost_item_id = $2 AND is_current = true`, + ownerID, trackID).Scan(&got); err != nil { + t.Fatalf("query repost: %v", err) + } + if !got { + t.Errorf("reposts.is_repost_of_repost = false, want true (the source row sets it; "+ + "metadata was %s)", txs[0].GetMetadata()) + } + }) + + // --- playlists.is_image_autogenerated / is_scheduled_release --------------- + t.Run("playlist_flags", func(t *testing.T) { + exec(src, `CREATE TABLE playlists ( + playlist_id bigint, playlist_owner_id bigint, playlist_name text, description text, + is_album boolean, is_private boolean, is_image_autogenerated boolean, + metadata_multihash text, playlist_image_sizes_multihash text, playlist_image_multihash text, + playlist_contents jsonb, upc text, ddex_app text, parental_warning_type text, + release_date timestamp, is_scheduled_release boolean, is_stream_gated boolean, + stream_conditions jsonb, ddex_release_ids jsonb, artists jsonb, + copyright_line jsonb, producer_copyright_line jsonb, + is_delete boolean, created_at timestamp, is_current boolean)`) + exec(src, `CREATE TABLE playlist_routes (playlist_id bigint, slug text, title_slug text, collision_id int, is_current boolean)`) + exec(src, `CREATE TABLE playlist_tracks (playlist_id bigint, track_id bigint, created_at timestamp, updated_at timestamp, is_removed boolean)`) + + insertPlaylist := `INSERT INTO playlists VALUES ( + $1, $2, $3, NULL, false, false, $4, + NULL, NULL, NULL, + '{"track_ids": []}'::jsonb, NULL, NULL, NULL, + $5, $6, false, + NULL, NULL, NULL, NULL, NULL, + false, $5, true)` + exec(src, insertPlaylist, playlistA, ownerID, "Autogen", true, createdAt, true) + // The negative control: a playlist with neither flag must stay false, so a + // blanket true would fail here rather than pass silently. + exec(src, insertPlaylist, playlistB, ownerID, "Plain", false, createdAt, false) + + w := newTestWriter(t, src) + if err := w.writePlaylists(ctx); err != nil { + t.Fatalf("writePlaylists: %v", err) + } + txs := decodeMigrationTxs(t, w.blockTxs, "Playlist", "Create") + if len(txs) != 2 { + t.Fatalf("emitted %d Playlist/Create transactions, want 2", len(txs)) + } + for _, tx := range txs { + replay(t, tx) + } + + for _, tc := range []struct { + id int64 + wantAutogen, wantSched bool + }{ + {playlistA, true, true}, + {playlistB, false, false}, + } { + var autogen, sched bool + if err := dst.QueryRow(ctx, + `SELECT is_image_autogenerated, is_scheduled_release FROM playlists WHERE playlist_id = $1 AND is_current = true`, + tc.id).Scan(&autogen, &sched); err != nil { + t.Fatalf("query playlist %d: %v", tc.id, err) + } + if autogen != tc.wantAutogen { + t.Errorf("playlist %d is_image_autogenerated = %v, want %v", tc.id, autogen, tc.wantAutogen) + } + if sched != tc.wantSched { + t.Errorf("playlist %d is_scheduled_release = %v, want %v", tc.id, sched, tc.wantSched) + } + } + }) + + // --- comments.is_members_only / video_url ---------------------------------- + t.Run("comment_fan_club_fields", func(t *testing.T) { + exec(src, `CREATE TABLE comments (comment_id bigint, text text, user_id bigint, + entity_id bigint, entity_type text, track_timestamp_s int, created_at timestamp, + is_delete boolean, is_members_only boolean, video_url text)`) + exec(src, `CREATE TABLE comment_threads (comment_id bigint, parent_comment_id bigint)`) + exec(src, `CREATE TABLE comment_mentions (comment_id bigint, user_id bigint, is_delete boolean)`) + + // is_members_only is only honored on entity_type='FanClub' — every one of + // the 39 source rows that sets it is a FanClub comment, and + // validateCommentWrite rejects the flag anywhere else. + exec(src, `INSERT INTO comments VALUES ($1, 'members only', $2, $3, 'FanClub', NULL, $4, false, true, $5)`, + commentA, ownerID, fanClubID, createdAt, videoURL) + // The negative control: a plain track comment must come back false/NULL. + exec(src, `INSERT INTO comments VALUES ($1, 'plain', $2, $3, 'Track', NULL, $4, false, false, NULL)`, + commentB, ownerID, trackID, createdAt) + + w := newTestWriter(t, src) + if err := w.writeComments(ctx); err != nil { + t.Fatalf("writeComments: %v", err) + } + txs := decodeMigrationTxs(t, w.blockTxs, "Comment", "Create") + if len(txs) != 2 { + t.Fatalf("emitted %d Comment/Create transactions, want 2", len(txs)) + } + for _, tx := range txs { + replay(t, tx) + } + + var membersOnly bool + var video *string + if err := dst.QueryRow(ctx, + `SELECT is_members_only, video_url FROM comments WHERE comment_id = $1`, + commentA).Scan(&membersOnly, &video); err != nil { + t.Fatalf("query comment %d: %v", commentA, err) + } + if !membersOnly { + t.Errorf("comment %d is_members_only = false, want true", commentA) + } + if video == nil || *video != videoURL { + t.Errorf("comment %d video_url = %v, want %q", commentA, video, videoURL) + } + + if err := dst.QueryRow(ctx, + `SELECT is_members_only, video_url FROM comments WHERE comment_id = $1`, + commentB).Scan(&membersOnly, &video); err != nil { + t.Fatalf("query comment %d: %v", commentB, err) + } + if membersOnly { + t.Errorf("comment %d is_members_only = true, want false", commentB) + } + if video != nil { + t.Errorf("comment %d video_url = %q, want NULL", commentB, *video) + } + }) +}