Skip to content

fix(genesis-writer): emit six state fields the indexer reads #491

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rickyrombo merged 1 commit into from
Aug 10, 2026
Merged

fix(genesis-writer): emit six state fields the indexer reads #491

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 13 additions & 2 deletions cmd/genesis-writer/entities_comment.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"`
Expand All @@ -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.
Expand Down Expand Up @@ -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+`
Expand All @@ -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 {
Expand All @@ -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.
Expand Down
66 changes: 36 additions & 30 deletions cmd/genesis-writer/entities_playlist.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"`
Expand Down Expand Up @@ -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 {
Expand All @@ -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,
Expand All @@ -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,
Expand All @@ -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),
Expand Down
48 changes: 37 additions & 11 deletions cmd/genesis-writer/entities_social.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
Expand Down Expand Up @@ -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)
}
Expand All @@ -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)
}
Expand Down
Loading
Loading