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 @@ "log" "strings" "time" + "github.com/bluesky-social/indigo/atproto/syntax" "tangled.org/core/appview/models" "tangled.org/core/orm" ) @@ -16,33 +17,29 @@ _, err := e.Exec(query, follow.UserDid, follow.SubjectDid, follow.Rkey) 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) +// Remove a follow +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, err + return nil, fmt.Errorf("deleting follows: %w", err) } + defer rows.Close() - 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 + 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 &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 + 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 @@ return &reaction, nil } // 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.Subject, 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 @@ "net/http" "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,30 +89,50 @@ }) 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 } - - _, err = comatproto.RepoDeleteRecord(r.Context(), client, &comatproto.RepoDeleteRecord_Input{ - Collection: tangled.GraphFollowNSID, - Repo: currentUser.Did, - Rkey: follow.Rkey, - }) + defer tx.Rollback() + follows, err := db.DeleteFollow(tx, syntax.DID(currentUser.Did), subjectIdent.DID) if err != nil { - l.Error("failed to unfollow", "err", err) + l.Error("failed to delete follows from db", "err", err) return } - err = db.DeleteFollowByRkey(s.db, currentUser.Did, 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.Warn("failed to delete follow from DB", "err", err) - // this is not an issue, the firehose event might have already done this + l.Error("failed to delete follows from PDS", "err", err) + return } + 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 { l.Error("failed to get follow stats", "err", err) @@ -122,8 +143,6 @@ UserDid: subjectIdent.DID.String(), 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 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 get reaction relationship", "did", currentUser.Did, "subjectUri", subjectUri, "err", err) + 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,