-63

spindle/db/member.go

Reviewed

··· 1 1 - package db 2 2 - 3 3 - import ( 4 4 - "github.com/bluesky-social/indigo/atproto/syntax" 5 5 - ) 6 6 - 7 7 - type SpindleMember struct { 8 8 - Id int 9 9 - Did syntax.DID // owner of the record 10 10 - Rkey string // rkey of the record 11 11 - Instance string 12 12 - Subject syntax.DID // the member being added 13 13 - } 14 14 - 15 15 - func AddSpindleMember(q DBTX, member SpindleMember) error { 16 16 - _, err := q.Exec( 17 17 - `insert or ignore into spindle_members (did, rkey, instance, subject) values (?, ?, ?, ?)`, 18 18 - member.Did, 19 19 - member.Rkey, 20 20 - member.Instance, 21 21 - member.Subject, 22 22 - ) 23 23 - return err 24 24 - } 25 25 - 26 26 - func RemoveSpindleMember(q DBTX, ownerDid, rkey string) error { 27 27 - _, err := q.Exec( 28 28 - "delete from spindle_members where did = ? and rkey = ?", 29 29 - ownerDid, 30 30 - rkey, 31 31 - ) 32 32 - return err 33 33 - } 34 34 - 35 35 - func CountSpindleMembersBySubject(q DBTX, subject string) (int, error) { 36 36 - var count int 37 37 - err := q.QueryRow( 38 38 - `select count(*) from spindle_members where subject = ?`, 39 39 - subject, 40 40 - ).Scan(&count) 41 41 - return count, err 42 42 - } 43 43 - 44 44 - func GetSpindleMember(q DBTX, did, rkey string) (*SpindleMember, error) { 45 45 - query := 46 46 - `select id, did, rkey, instance, subject 47 47 - from spindle_members 48 48 - where did = ? and rkey = ?` 49 49 - 50 50 - var member SpindleMember 51 51 - err := q.QueryRow(query, did, rkey).Scan( 52 52 - &member.Id, 53 53 - &member.Did, 54 54 - &member.Rkey, 55 55 - &member.Instance, 56 56 - &member.Subject, 57 57 - ) 58 58 - if err != nil { 59 59 - return nil, err 60 60 - } 61 61 - 62 62 - return &member, nil 63 63 - }

-188

spindle/ingester.go

View file Reviewed

··· 2 2 3 3 import ( 4 4 "context" 5 5 - "database/sql" 6 6 - "encoding/json" 7 7 - "errors" 8 8 - "fmt" 9 5 10 6 "tangled.org/core/api/tangled" 11 11 - "tangled.org/core/spindle/db" 12 7 "tangled.org/core/tapc" 13 8 14 9 "github.com/bluesky-social/indigo/atproto/syntax" ··· 25 20 26 21 var err error 27 22 switch e.Commit.Collection { 28 28 - case tangled.SpindleMemberNSID: 29 29 - err = s.ingestMember(ctx, e) 30 23 case tangled.RepoNSID, tangled.RepoCollaboratorNSID: 31 24 if evt, ok := jetstreamToTapEvent(e); ok { 32 25 err = s.tap.processEvent(ctx, evt) ··· 77 70 }, 78 71 }, true 79 72 } 80 80 - 81 81 - func (s *Spindle) ingestMember(ctx context.Context, e *models.Event) error { 82 82 - did := e.Did 83 83 - rkey := e.Commit.RKey 84 84 - l := s.l.With("component", "ingester", "record", tangled.SpindleMemberNSID, "did", did, "rkey", rkey) 85 85 - 86 86 - switch e.Commit.Operation { 87 87 - case models.CommitOperationCreate, models.CommitOperationUpdate: 88 88 - raw := e.Commit.Record 89 89 - record := tangled.SpindleMember{} 90 90 - if err := json.Unmarshal(raw, &record); err != nil { 91 91 - return fmt.Errorf("invalid record: %w", err) 92 92 - } 93 93 - 94 94 - domain := s.cfg.Server.Hostname 95 95 - recordInstance := record.Instance 96 96 - 97 97 - if recordInstance != domain { 98 98 - return fmt.Errorf("domain mismatch: %s != %s", record.Instance, domain) 99 99 - } 100 100 - 101 101 - subject, err := syntax.ParseDID(record.Subject) 102 102 - if err != nil { 103 103 - return fmt.Errorf("invalid subject DID %q: %w", record.Subject, err) 104 104 - } 105 105 - 106 106 - ok, err := s.e.IsSpindleInviteAllowed(did, rbacDomain) 107 107 - if err != nil { 108 108 - return fmt.Errorf("failed to enforce permissions: %w", err) 109 109 - } 110 110 - if !ok { 111 111 - return fmt.Errorf("permission denied for %s", did) 112 112 - } 113 113 - 114 114 - sqlTx, err := s.db.BeginTx(ctx, nil) 115 115 - if err != nil { 116 116 - return fmt.Errorf("failed to start txn: %w", err) 117 117 - } 118 118 - committed := false 119 119 - defer func() { 120 120 - if !committed { 121 121 - sqlTx.Rollback() 122 122 - } 123 123 - }() 124 124 - 125 125 - existing, err := db.GetSpindleMember(sqlTx, did, rkey) 126 126 - if err != nil && !errors.Is(err, sql.ErrNoRows) { 127 127 - return fmt.Errorf("failed to look up existing member: %w", err) 128 128 - } 129 129 - 130 130 - var staleSubject string 131 131 - if existing != nil && existing.Subject != subject { 132 132 - staleSubject = existing.Subject.String() 133 133 - if err := db.RemoveSpindleMember(sqlTx, did, rkey); err != nil { 134 134 - return fmt.Errorf("failed to remove stale member row: %w", err) 135 135 - } 136 136 - } 137 137 - 138 138 - if err := db.AddSpindleMember(sqlTx, db.SpindleMember{ 139 139 - Did: syntax.DID(did), 140 140 - Rkey: rkey, 141 141 - Instance: recordInstance, 142 142 - Subject: subject, 143 143 - }); err != nil { 144 144 - return fmt.Errorf("failed to add member: %w", err) 145 145 - } 146 146 - 147 147 - if err := db.AddDid(sqlTx, subject.String()); err != nil { 148 148 - return fmt.Errorf("failed to add did: %w", err) 149 149 - } 150 150 - 151 151 - dropStaleAcl := false 152 152 - var staleDidDropped bool 153 153 - if staleSubject != "" { 154 154 - remaining, err := db.CountSpindleMembersBySubject(sqlTx, staleSubject) 155 155 - if err != nil { 156 156 - return fmt.Errorf("failed to count stale subject rows: %w", err) 157 157 - } 158 158 - if remaining == 0 { 159 159 - dropStaleAcl = true 160 160 - stillNeeded, err := s.e.WouldHaveAnyPolicyExcludingSpindleMember(staleSubject, rbacDomain) 161 161 - if err != nil { 162 162 - return fmt.Errorf("failed to check residual policies for stale subject: %w", err) 163 163 - } 164 164 - if !stillNeeded { 165 165 - if err := db.RemoveDid(sqlTx, staleSubject); err != nil { 166 166 - return fmt.Errorf("failed to remove stale did: %w", err) 167 167 - } 168 168 - staleDidDropped = true 169 169 - } 170 170 - } 171 171 - l.Info("replaced stale spindle member", "old_subject", staleSubject, "new_subject", subject, "stale_did_dropped", staleDidDropped) 172 172 - } 173 173 - 174 174 - if err := sqlTx.Commit(); err != nil { 175 175 - return fmt.Errorf("failed to commit txn: %w", err) 176 176 - } 177 177 - committed = true 178 178 - 179 179 - if dropStaleAcl { 180 180 - if _, err := s.e.TryRemoveSpindleMember(rbacDomain, staleSubject); err != nil { 181 181 - l.Error("post-commit: failed to remove stale ACL", "subject", staleSubject, "err", err) 182 182 - } 183 183 - } 184 184 - if _, err := s.e.TryAddSpindleMember(rbacDomain, subject.String()); err != nil { 185 185 - l.Error("post-commit: failed to add member ACL", "subject", subject, "err", err) 186 186 - } 187 187 - 188 188 - if staleDidDropped { 189 189 - s.jc.RemoveDid(staleSubject) 190 190 - } 191 191 - s.jc.AddDid(subject.String()) 192 192 - l.Info("added member from firehose", "member", subject) 193 193 - return nil 194 194 - 195 195 - case models.CommitOperationDelete: 196 196 - sqlTx, err := s.db.BeginTx(ctx, nil) 197 197 - if err != nil { 198 198 - return fmt.Errorf("failed to start txn: %w", err) 199 199 - } 200 200 - committed := false 201 201 - defer func() { 202 202 - if !committed { 203 203 - sqlTx.Rollback() 204 204 - } 205 205 - }() 206 206 - 207 207 - record, err := db.GetSpindleMember(sqlTx, did, rkey) 208 208 - if errors.Is(err, sql.ErrNoRows) { 209 209 - l.Info("spindle member already removed") 210 210 - return nil 211 211 - } 212 212 - if err != nil { 213 213 - return fmt.Errorf("failed to find member: %w", err) 214 214 - } 215 215 - 216 216 - staleSubject := record.Subject.String() 217 217 - 218 218 - if err := db.RemoveSpindleMember(sqlTx, did, rkey); err != nil { 219 219 - return fmt.Errorf("failed to remove member: %w", err) 220 220 - } 221 221 - 222 222 - remaining, err := db.CountSpindleMembersBySubject(sqlTx, staleSubject) 223 223 - if err != nil { 224 224 - return fmt.Errorf("failed to count remaining member rows: %w", err) 225 225 - } 226 226 - 227 227 - dropAcl := false 228 228 - var staleDidDropped bool 229 229 - if remaining == 0 { 230 230 - dropAcl = true 231 231 - stillNeeded, err := s.e.WouldHaveAnyPolicyExcludingSpindleMember(staleSubject, rbacDomain) 232 232 - if err != nil { 233 233 - return fmt.Errorf("failed to check residual policies: %w", err) 234 234 - } 235 235 - if !stillNeeded { 236 236 - if err := db.RemoveDid(sqlTx, staleSubject); err != nil { 237 237 - return fmt.Errorf("failed to remove did: %w", err) 238 238 - } 239 239 - staleDidDropped = true 240 240 - } 241 241 - } 242 242 - 243 243 - if err := sqlTx.Commit(); err != nil { 244 244 - return fmt.Errorf("failed to commit txn: %w", err) 245 245 - } 246 246 - committed = true 247 247 - 248 248 - if dropAcl { 249 249 - if _, err := s.e.TryRemoveSpindleMember(rbacDomain, staleSubject); err != nil { 250 250 - l.Error("post-commit: failed to remove member ACL", "subject", staleSubject, "err", err) 251 251 - } 252 252 - } 253 253 - 254 254 - if staleDidDropped { 255 255 - s.jc.RemoveDid(staleSubject) 256 256 - } 257 257 - l.Info("removed member from firehose", "member", record.Subject, "remaining_rows", remaining, "stale_did_dropped", staleDidDropped) 258 258 - } 259 259 - return nil 260 260 - }