From 8684a37d576fd98ddb92061136ab1c0170eec93a Mon Sep 17 00:00:00 2001 From: Seongmin Lee Date: Mon, 19 Jan 2026 15:15:20 +0000 Subject: [PATCH] appview/{db,state}: delete all duplicate records on delete Signed-off-by: Seongmin Lee --- appview/db/follow.go | 49 +++++++++++++++++++++++-------------------------- appview/db/reaction.go | 26 +++++++++++++++++++++++--- appview/db/star.go | 57 ++++++++++++++++++++++++++------------------------------- appview/state/follow.go | 49 ++++++++++++++++++++++++++++++++++--------------- appview/state/reaction.go | 36 ++++++++++++++++++++++++------------ appview/state/star.go | 47 ++++++++++++++++++++++++++++++++--------------- 6 file(s) changed, 162 insertion(s)(+), 102 deletion(s)(-) diff --git a/appview/db/follow.go b/appview/db/follow.go --- a/appview/db/follow.go +++ b/appview/db/follow.go @@ -6,6 +6,7 @@ "strings" "time" + "github.com/bluesky-social/indigo/atproto/syntax" "tangled.org/core/appview/models" "tangled.org/core/orm" ) @@ -16,33 +17,29 @@ return err } -// Get a follow record -func GetFollow(e Execer, userDid, subjectDid string) (*models.Follow, error) { - query := `select did, subject_did, created, rkey from follows where did = ? and subject_did = ?` - row := e.QueryRow(query, userDid, subjectDid) - - var follow models.Follow - var followedAt string - err := row.Scan(&follow.UserDid, &follow.SubjectDid, &followedAt, &follow.Rkey) - if err != nil { - return nil, err - } - - followedAtTime, err := time.Parse(time.RFC3339, followedAt) - if err != nil { - log.Println("unable to determine followed at time") - follow.FollowedAt = time.Now() - } else { - follow.FollowedAt = followedAtTime - } - - return &follow, nil -} - // Remove a follow -func DeleteFollow(e Execer, userDid, subjectDid string) error { - _, err := e.Exec(`delete from follows where did = ? and subject_did = ?`, userDid, subjectDid) - return err +func DeleteFollow(e Execer, did, subjectDid syntax.DID) ([]syntax.ATURI, error) { + var deleted []syntax.ATURI + rows, err := e.Query( + `delete from follows + where did = ? and subject_did = ? + returning at_uri`, + did, + subjectDid, + ) + if err != nil { + return nil, fmt.Errorf("deleting follows: %w", err) + } + defer rows.Close() + + for rows.Next() { + var aturi syntax.ATURI + if err := rows.Scan(&aturi); err != nil { + return nil, fmt.Errorf("scanning at_uri: %w", err) + } + deleted = append(deleted, aturi) + } + return deleted, nil } // Remove a follow diff --git a/appview/db/reaction.go b/appview/db/reaction.go --- a/appview/db/reaction.go +++ b/appview/db/reaction.go @@ -43,9 +43,29 @@ } // Remove a reaction -func DeleteReaction(e Execer, did string, subjectAt syntax.ATURI, kind models.ReactionKind) error { - _, err := e.Exec(`delete from reactions where did = ? and subject_at = ? and kind = ?`, did, subjectAt, kind) - return err +func DeleteReaction(e Execer, did syntax.DID, subjectAt syntax.ATURI, kind models.ReactionKind) ([]syntax.ATURI, error) { + var deleted []syntax.ATURI + rows, err := e.Query( + `delete from reactions + where did = ? and subject_at = ? and kind = ? + returning at_uri`, + did, + subjectAt, + kind, + ) + if err != nil { + return nil, fmt.Errorf("deleting stars: %w", err) + } + defer rows.Close() + + for rows.Next() { + var aturi syntax.ATURI + if err := rows.Scan(&aturi); err != nil { + return nil, fmt.Errorf("scanning at_uri: %w", err) + } + deleted = append(deleted, aturi) + } + return deleted, nil } // Remove a reaction diff --git a/appview/db/star.go b/appview/db/star.go --- a/appview/db/star.go +++ b/appview/db/star.go @@ -1,12 +1,13 @@ package db import ( + "database/sql" "fmt" - "log" "slices" "strings" "time" + "github.com/bluesky-social/indigo/atproto/syntax" "tangled.org/core/appview/models" "tangled.org/core/appview/pagination" "tangled.org/core/orm" @@ -22,32 +23,6 @@ star.Rkey, ) return err -} - -// Get a star record -func GetStar(e Execer, did string, subject string) (*models.Star, error) { - query := ` - select did, subject_type, subject, created, rkey - from stars - where did = ? and subject = ?` - row := e.QueryRow(query, did, subject) - - var star models.Star - var created string - err := row.Scan(&star.Did, &star.SubjectType, &star.Subject, &created, &star.Rkey) - if err != nil { - return nil, err - } - - createdAtTime, err := time.Parse(time.RFC3339, created) - if err != nil { - log.Println("unable to determine followed at time") - star.Created = time.Now() - } else { - star.Created = createdAtTime - } - - return &star, nil } func GetStars(e Execer, subject string, page pagination.Page) ([]models.Star, error) { @@ -82,10 +57,30 @@ return stars, rows.Err() } -// Remove a star -func DeleteStar(e Execer, did string, subject string) error { - _, err := e.Exec(`delete from stars where did = ? and subject = ?`, did, subject) - return err +// Remove all stars from given user to subject +func DeleteStars(tx *sql.Tx, did syntax.DID, subject string) ([]syntax.ATURI, error) { + var deleted []syntax.ATURI + rows, err := tx.Query( + `delete from stars + where did = ? and subject = ? + returning at_uri`, + did, + subject, + ) + if err != nil { + return nil, fmt.Errorf("deleting stars: %w", err) + } + defer rows.Close() + + for rows.Next() { + var aturi syntax.ATURI + if err := rows.Scan(&aturi); err != nil { + return nil, fmt.Errorf("scanning at_uri: %w", err) + } + deleted = append(deleted, aturi) + } + + return deleted, nil } // Remove a star diff --git a/appview/state/follow.go b/appview/state/follow.go --- a/appview/state/follow.go +++ b/appview/state/follow.go @@ -5,6 +5,7 @@ "time" comatproto "github.com/bluesky-social/indigo/api/atproto" + "github.com/bluesky-social/indigo/atproto/syntax" lexutil "github.com/bluesky-social/indigo/lex/util" "tangled.org/core/api/tangled" "tangled.org/core/appview/db" @@ -88,29 +89,49 @@ return case http.MethodDelete: - // find the record in the db - follow, err := db.GetFollow(s.db, currentUser.Did, subjectIdent.DID.String()) + tx, err := s.db.BeginTx(r.Context(), nil) if err != nil { - l.Error("failed to get follow relationship", "err", err) + l.Error("failed to start transaction", "err", err) + return + } + defer tx.Rollback() + + follows, err := db.DeleteFollow(tx, syntax.DID(currentUser.Did), subjectIdent.DID) + if err != nil { + l.Error("failed to delete follows from db", "err", err) return } - _, err = comatproto.RepoDeleteRecord(r.Context(), client, &comatproto.RepoDeleteRecord_Input{ - Collection: tangled.GraphFollowNSID, - Repo: currentUser.Did, - Rkey: follow.Rkey, + var writes []*comatproto.RepoApplyWrites_Input_Writes_Elem + for _, followAt := range follows { + writes = append(writes, &comatproto.RepoApplyWrites_Input_Writes_Elem{ + RepoApplyWrites_Delete: &comatproto.RepoApplyWrites_Delete{ + Collection: tangled.GraphFollowNSID, + Rkey: followAt.RecordKey().String(), + }, + }) + } + _, err = comatproto.RepoApplyWrites(r.Context(), client, &comatproto.RepoApplyWrites_Input{ + Repo: currentUser.Did, + Writes: writes, }) - if err != nil { - l.Error("failed to unfollow", "err", err) + l.Error("failed to delete follows from PDS", "err", err) return } - err = db.DeleteFollowByRkey(s.db, currentUser.Did, follow.Rkey) - if err != nil { - l.Warn("failed to delete follow from DB", "err", err) - // this is not an issue, the firehose event might have already done this + if err := tx.Commit(); err != nil { + l.Error("failed to commit transaction", "err", err) + // The record was deleted from the PDS but the local rollback kept it. + // Ingester will backfill the missed operation } + + s.notifier.DeleteFollow(r.Context(), &models.Follow{ + UserDid: currentUser.Did, + SubjectDid: subjectIdent.DID.String(), + // Rkey + // FollowedAt + }) followStats, err := db.GetFollowerFollowingCount(s.db, subjectIdent.DID.String()) if err != nil { @@ -122,8 +143,6 @@ FollowStatus: models.IsNotFollowing, FollowersCount: followStats.Followers, }) - - s.notifier.DeleteFollow(r.Context(), follow) return } diff --git a/appview/state/reaction.go b/appview/state/reaction.go --- a/appview/state/reaction.go +++ b/appview/state/reaction.go @@ -91,27 +91,39 @@ return case http.MethodDelete: - reaction, err := db.GetReaction(s.db, currentUser.Did, subjectUri, reactionKind) + tx, err := s.db.BeginTx(r.Context(), nil) if err != nil { - l.Error("failed to get reaction relationship", "did", currentUser.Did, "subjectUri", subjectUri, "err", err) + l.Error("failed to start transaction", "err", err) + } + defer tx.Rollback() + + reactions, err := db.DeleteReaction(tx, syntax.DID(currentUser.Did), subjectUri, reactionKind) + if err != nil { + l.Error("failed to delete reactions from db", "err", err) return } - _, err = comatproto.RepoDeleteRecord(r.Context(), client, &comatproto.RepoDeleteRecord_Input{ - Collection: tangled.FeedReactionNSID, - Repo: currentUser.Did, - Rkey: reaction.Rkey, + var writes []*comatproto.RepoApplyWrites_Input_Writes_Elem + for _, reactionAt := range reactions { + writes = append(writes, &comatproto.RepoApplyWrites_Input_Writes_Elem{ + RepoApplyWrites_Delete: &comatproto.RepoApplyWrites_Delete{ + Collection: tangled.FeedReactionNSID, + Rkey: reactionAt.RecordKey().String(), + }, + }) + } + _, err = comatproto.RepoApplyWrites(r.Context(), client, &comatproto.RepoApplyWrites_Input{ + Repo: currentUser.Did, + Writes: writes, }) - if err != nil { - l.Error("failed to remove reaction", "err", err) + l.Error("failed to delete reactions from PDS", "err", err) return } - err = db.DeleteReactionByRkey(s.db, currentUser.Did, reaction.Rkey) - if err != nil { - l.Warn("failed to delete reaction from DB", "err", err) - // this is not an issue, the firehose event might have already done this + if err := tx.Commit(); err != nil { + l.Error("failed to commit transaction", "err", err) + // DB op failed but record is created in PDS. Ingester will backfill the missed operation } reactionMap, err := db.GetReactionMap(s.db, 20, subjectUri) diff --git a/appview/state/star.go b/appview/state/star.go --- a/appview/state/star.go +++ b/appview/state/star.go @@ -126,37 +126,54 @@ return case http.MethodDelete: - // find the record in the db - star, err := db.GetStar(s.db, currentUser.Did, subjectKey) + tx, err := s.db.BeginTx(r.Context(), nil) if err != nil { - l.Error("failed to get star relationship", "err", err) + l.Error("failed to start transaction", "err", err) + } + defer tx.Rollback() + + stars, err := db.DeleteStars(tx, syntax.DID(currentUser.Did), subjectKey) + if err != nil { + l.Error("failed to delete stars from db", "err", err) return } - _, err = comatproto.RepoDeleteRecord(r.Context(), client, &comatproto.RepoDeleteRecord_Input{ - Collection: tangled.FeedStarNSID, - Repo: currentUser.Did, - Rkey: star.Rkey, + var writes []*comatproto.RepoApplyWrites_Input_Writes_Elem + for _, starAt := range stars { + writes = append(writes, &comatproto.RepoApplyWrites_Input_Writes_Elem{ + RepoApplyWrites_Delete: &comatproto.RepoApplyWrites_Delete{ + Collection: tangled.FeedStarNSID, + Rkey: starAt.RecordKey().String(), + }, + }) + } + _, err = comatproto.RepoApplyWrites(r.Context(), client, &comatproto.RepoApplyWrites_Input{ + Repo: currentUser.Did, + Writes: writes, }) - if err != nil { - l.Error("failed to unstar", "err", err) + l.Error("failed to delete stars from PDS", "err", err) return } - err = db.DeleteStarByRkey(s.db, currentUser.Did, star.Rkey) - if err != nil { - l.Warn("failed to delete star from DB", "err", err) - // this is not an issue, the firehose event might have already done this + if err := tx.Commit(); err != nil { + l.Error("failed to commit transaction", "err", err) + // DB op failed but record is created in PDS. Ingester will backfill the missed operation } + + s.notifier.DeleteStar(r.Context(), &models.Star{ + Did: currentUser.Did, + SubjectType: subjectType, + Subject: subjectKey, + // Rkey + // Created + }) starCount, err := db.GetStarCount(s.db, subjectType, subjectKey) if err != nil { l.Error("failed to get star count", "subject", subjectKey, "err", err) return } - - s.notifier.DeleteStar(r.Context(), star) s.pages.StarBtnFragment(w, pages.StarBtnFragmentParams{ IsStarred: false, -- tangled.sh