appview/ingester: ingest label op crud via parking · beathagenlocher.com/core@34096ce

+16

appview/db/label.go

··· 223 223 return id, nil 224 224 } 225 225 226 + func DeleteLabelOps(e Execer, filters ...orm.Filter) error { 227 + var conditions []string 228 + var args []any 229 + for _, filter := range filters { 230 + conditions = append(conditions, filter.Condition()) 231 + args = append(args, filter.Arg()...) 232 + } 233 + whereClause := "" 234 + if conditions != nil { 235 + whereClause = " where " + strings.Join(conditions, " and ") 236 + } 237 + query := fmt.Sprintf(`delete from label_ops %s`, whereClause) 238 + _, err := e.Exec(query, args...) 239 + return err 240 + } 241 + 226 242 func GetLabelOps(e Execer, filters ...orm.Filter) ([]models.LabelOp, error) { 227 243 var labelOps []models.LabelOp 228 244 var conditions []string

+153 -91

appview/ingester.go

··· 8 8 "fmt" 9 9 "io" 10 10 "log/slog" 11 - "maps" 12 11 "net/http" 13 12 "net/url" 14 13 "slices" ··· 1409 1408 return err 1410 1409 } 1411 1410 1412 - i.drainPendingState(ctx, issue.AtUri(), issueStateSpec, l) 1411 + i.drainPendingState(ctx, issue.AtUri(), l) 1413 1412 1414 1413 l.Info("ingested record") 1415 1414 return nil ··· 1576 1575 return err 1577 1576 } 1578 1577 1579 - i.drainPendingState(ctx, pull.AtUri(), pullStatusSpec, l) 1578 + i.drainPendingState(ctx, pull.AtUri(), l) 1580 1579 1581 1580 l.Info("ingested record") 1582 1581 return nil ··· 1815 1814 return err 1816 1815 } 1817 1816 1818 - l.Info("parked state record for retry", "subject", subject) 1817 + l.Info("parked record for retry", "subject", subject, "nsid", nsid) 1819 1818 return nil 1820 1819 } 1821 1820 1822 - func (i *Ingester) drainPendingState(ctx context.Context, subject syntax.ATURI, spec stateIngestSpec, l *slog.Logger) { 1821 + func (i *Ingester) drainPendingState(ctx context.Context, subject syntax.ATURI, l *slog.Logger) { 1823 1822 pending, err := db.PendingStateRecordsForSubject(i.Db, subject) 1824 1823 if err != nil { 1825 - l.Error("failed to load pending state records", "err", err, "subject", subject) 1824 + l.Error("failed to load pending records", "err", err, "subject", subject) 1826 1825 return 1827 1826 } 1828 1827 for _, p := range pending { 1829 - if err := i.applyStateRecord(ctx, p.Did, p.Rkey, p.Nsid, p.Record, spec, l); err != nil { 1830 - l.Error("failed to drain pending state record", "err", err, "did", p.Did, "rkey", p.Rkey) 1828 + if err := i.reapplyPendingRecord(ctx, p, l); err != nil { 1829 + l.Error("failed to drain pending record", "err", err, "did", p.Did, "rkey", p.Rkey, "nsid", p.Nsid) 1831 1830 } 1832 1831 } 1833 1832 } 1834 1833 1834 + func (i *Ingester) drainPendingLabelOps(l *slog.Logger) { 1835 + subjects, err := db.PendingStateSubjectsForNsid(i.Db, tangled.LabelOpNSID) 1836 + if err != nil { 1837 + l.Error("failed to list pending label op subjects", "err", err) 1838 + return 1839 + } 1840 + for _, subject := range subjects { 1841 + i.drainPendingState(i.Ctx, subject, l) 1842 + } 1843 + } 1844 + 1845 + func (i *Ingester) reapplyPendingRecord(ctx context.Context, p db.PendingStateRecord, l *slog.Logger) error { 1846 + switch p.Nsid { 1847 + case tangled.RepoIssueStateNSID: 1848 + return i.applyStateRecord(ctx, p.Did, p.Rkey, p.Nsid, p.Record, issueStateSpec, l) 1849 + case tangled.RepoPullStatusNSID: 1850 + return i.applyStateRecord(ctx, p.Did, p.Rkey, p.Nsid, p.Record, pullStatusSpec, l) 1851 + case tangled.LabelOpNSID: 1852 + return i.applyLabelOpRecord(ctx, p.Did, p.Rkey, p.Record, l) 1853 + default: 1854 + return fmt.Errorf("no reapply handler for parked nsid: %s", p.Nsid) 1855 + } 1856 + } 1857 + 1835 1858 const ( 1836 1859 pendingStateReconcileInterval = time.Hour 1837 1860 pendingStateRecordTTL = 7 * 24 * time.Hour 1838 1861 ) 1839 - 1840 - func stateSpecForSubject(subject syntax.ATURI) (stateIngestSpec, bool) { 1841 - switch string(subject.Collection()) { 1842 - case tangled.RepoIssueNSID: 1843 - return issueStateSpec, true 1844 - case tangled.RepoPullNSID: 1845 - return pullStatusSpec, true 1846 - default: 1847 - return stateIngestSpec{}, false 1848 - } 1849 - } 1850 1862 1851 1863 func (i *Ingester) StartPendingStateReconciler() { 1852 1864 i.ReconcilePendingState() ··· 1871 1883 l.Error("failed to list pending state subjects", "err", err) 1872 1884 } 1873 1885 for _, subject := range subjects { 1874 - spec, ok := stateSpecForSubject(subject) 1875 - if !ok { 1876 - continue 1877 - } 1878 - i.drainPendingState(i.Ctx, subject, spec, l) 1886 + i.drainPendingState(i.Ctx, subject, l) 1879 1887 } 1880 1888 1881 1889 cutoff := time.Now().Add(-pendingStateRecordTTL).UTC().Format(time.RFC3339) ··· 2081 2089 return fmt.Errorf("failed to create labeldef: %w", err) 2082 2090 } 2083 2091 2092 + if e.Commit.Operation == jmodels.CommitOperationCreate { 2093 + i.drainPendingLabelOps(l) 2094 + } 2095 + 2084 2096 l.Info("ingested record") 2085 2097 return nil 2086 2098 ··· 2101 2113 } 2102 2114 2103 2115 func (i *Ingester) ingestLabelOp(ctx context.Context, e *jmodels.Event, l *slog.Logger) error { 2116 + l = l.With("handler", "ingestLabelOp") 2104 2117 did := e.Did 2105 2118 rkey := e.Commit.RKey 2106 2119 2107 - var err error 2120 + switch e.Commit.Operation { 2121 + case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: 2122 + return i.applyLabelOpRecord(ctx, did, rkey, e.Commit.Record, l) 2123 + case jmodels.CommitOperationDelete: 2124 + return i.deleteLabelOpRecord(ctx, did, rkey, l) 2125 + } 2108 2126 2109 - l = l.With("handler", "ingestLabelOp") 2127 + return nil 2128 + } 2129 + 2130 + func (i *Ingester) findLabelSubjectRepo(subject syntax.ATURI) (*models.Repo, bool, error) { 2131 + var spec stateIngestSpec 2132 + switch subject.Collection() { 2133 + case tangled.RepoIssueNSID: 2134 + spec = issueStateSpec 2135 + case tangled.RepoPullNSID: 2136 + spec = pullStatusSpec 2137 + default: 2138 + return nil, false, fmt.Errorf("unsupported label subject: %s", subject.Collection()) 2139 + } 2140 + repo, _, found, err := spec.findSubject(i.Db, subject) 2141 + return repo, found, err 2142 + } 2143 + 2144 + func (i *Ingester) applyLabelOpRecord(ctx context.Context, did, rkey string, raw []byte, l *slog.Logger) error { 2145 + record := tangled.LabelOp{} 2146 + if err := json.Unmarshal(raw, &record); err != nil { 2147 + return fmt.Errorf("invalid record: %w", err) 2148 + } 2110 2149 2111 - switch e.Commit.Operation { 2112 - case jmodels.CommitOperationCreate: 2113 - raw := json.RawMessage(e.Commit.Record) 2114 - record := tangled.LabelOp{} 2115 - err = json.Unmarshal(raw, &record) 2116 - if err != nil { 2117 - return fmt.Errorf("invalid record: %w", err) 2118 - } 2150 + subject := syntax.ATURI(record.Subject) 2151 + park := func() error { 2152 + return i.parkStateRecord(ctx, did, rkey, tangled.LabelOpNSID, subject, raw, l) 2153 + } 2119 2154 2120 - subject := syntax.ATURI(record.Subject) 2121 - collection := subject.Collection() 2155 + repo, found, err := i.findLabelSubjectRepo(subject) 2156 + if err != nil { 2157 + return err 2158 + } 2159 + if !found { 2160 + return park() 2161 + } 2122 2162 2123 - var repo *models.Repo 2124 - switch collection { 2125 - case tangled.RepoIssueNSID: 2126 - i, err := db.GetIssues(i.Db, orm.FilterEq("at_uri", subject)) 2127 - if err != nil || len(i) != 1 { 2128 - return fmt.Errorf("failed to find subject: %w || subject count %d", err, len(i)) 2129 - } 2130 - repo = i[0].Repo 2131 - case tangled.RepoPullNSID: 2132 - p, err := db.GetPulls(i.Db, orm.FilterEq("at_uri", subject)) 2133 - if err != nil || len(p) != 1 { 2134 - return fmt.Errorf("failed to find subject: %w || subject count %d", err, len(p)) 2135 - } 2136 - repo = p[0].Repo 2137 - default: 2138 - return fmt.Errorf("unsupported label subject: %s", collection) 2163 + // validate permissions: only collaborators can apply labels currently 2164 + // 2165 + // TODO: introduce a repo:triage permission 2166 + allowed, permErr := i.Acl.HasRepoPermissionErr(ctx, repo, did, "repo:push") 2167 + if permErr != nil { 2168 + if !errors.Is(permErr, knotacl.ErrKnotUnreachable) { 2169 + return park() 2139 2170 } 2171 + l.Warn("ingesting labelop without permission check", "did", did, "err", permErr) 2172 + allowed = true 2173 + } 2140 2174 2141 - actx, err := db.NewLabelApplicationCtx(i.Db, orm.FilterIn("at_uri", repo.Labels)) 2142 - if err != nil { 2143 - return fmt.Errorf("failed to build label application ctx: %w", err) 2175 + if !allowed { 2176 + if err := i.unparkLabelOp(ctx, did, rkey); err != nil { 2177 + return err 2144 2178 } 2179 + l.Warn("dropped unauthorized label op", "did", did, "rkey", rkey, "subject", subject) 2180 + return nil 2181 + } 2145 2182 2146 - ops := models.LabelOpsFromRecord(did, rkey, record) 2183 + ops := models.LabelOpsFromRecord(did, rkey, record) 2184 + 2185 + operandKeys := make([]string, 0, len(ops)) 2186 + for idx := range ops { 2187 + operandKeys = append(operandKeys, ops[idx].OperandKey) 2188 + } 2147 2189 2148 - for _, o := range ops { 2149 - def, ok := actx.Defs[o.OperandKey] 2150 - if !ok { 2151 - return fmt.Errorf("failed to find label def for key: %s, expected: %q", o.OperandKey, slices.Collect(maps.Keys(actx.Defs))) 2152 - } 2153 - // validate permissions: only collaborators can apply labels currently 2154 - // 2155 - // TODO: introduce a repo:triage permission 2156 - allowed, permErr := i.Acl.HasRepoPermissionErr(ctx, repo, o.Did, "repo:push") 2157 - if permErr != nil { 2158 - if !errors.Is(permErr, knotacl.ErrKnotUnreachable) { 2159 - return fmt.Errorf("enforcing permission: %w", permErr) 2160 - } 2161 - l.Warn("ingesting labelop without permission check", "did", o.Did, "err", permErr) 2162 - } else if !allowed { 2163 - return fmt.Errorf("unauthorized label operation") 2164 - } 2190 + actx, err := db.NewLabelApplicationCtx(i.Db, orm.FilterIn("at_uri", operandKeys)) 2191 + if err != nil { 2192 + return fmt.Errorf("failed to build label application ctx: %w", err) 2193 + } 2165 2194 2166 - if err := def.ValidateOperandValue(&o); err != nil { 2167 - return fmt.Errorf("failed to validate labelop: %w", err) 2168 - } 2195 + for idx := range ops { 2196 + def, ok := actx.Defs[ops[idx].OperandKey] 2197 + if !ok { 2198 + return park() 2199 + } 2200 + if err := def.ValidateOperandValue(&ops[idx]); err != nil { 2201 + return fmt.Errorf("failed to validate labelop: %w", err) 2169 2202 } 2203 + } 2170 2204 2171 - tx, err := i.Db.Begin() 2172 - if err != nil { 2173 - return err 2174 - } 2175 - defer tx.Rollback() 2205 + if err := i.materializeLabelOps(ctx, did, rkey, ops); err != nil { 2206 + return err 2207 + } 2176 2208 2177 - for _, o := range ops { 2178 - _, err = db.AddLabelOp(tx, &o) 2179 - if err != nil { 2180 - return fmt.Errorf("failed to add labelop: %w", err) 2181 - } 2182 - } 2209 + l.Info("ingested record") 2210 + return nil 2211 + } 2183 2212 2184 - if err = tx.Commit(); err != nil { 2185 - return err 2186 - } 2213 + func (i *Ingester) unparkLabelOp(ctx context.Context, did, rkey string) error { 2214 + tx, err := i.Db.BeginTx(ctx, nil) 2215 + if err != nil { 2216 + return err 2217 + } 2218 + defer tx.Rollback() 2187 2219 2188 - l.Info("ingested record") 2220 + if err := db.UnparkStateRecord(tx, did, rkey, tangled.LabelOpNSID); err != nil { 2221 + return fmt.Errorf("failed to unpark label op: %w", err) 2189 2222 } 2190 2223 2224 + return tx.Commit() 2225 + } 2226 + 2227 + func (i *Ingester) materializeLabelOps(ctx context.Context, did, rkey string, ops []models.LabelOp) error { 2228 + tx, err := i.Db.BeginTx(ctx, nil) 2229 + if err != nil { 2230 + return err 2231 + } 2232 + defer tx.Rollback() 2233 + 2234 + if err := db.UnparkStateRecord(tx, did, rkey, tangled.LabelOpNSID); err != nil { 2235 + return fmt.Errorf("failed to unpark label op: %w", err) 2236 + } 2237 + if err := db.DeleteLabelOps(tx, orm.FilterEq("did", did), orm.FilterEq("rkey", rkey)); err != nil { 2238 + return fmt.Errorf("failed to clear prior label ops: %w", err) 2239 + } 2240 + for idx := range ops { 2241 + if _, err := db.AddLabelOp(tx, &ops[idx]); err != nil { 2242 + return fmt.Errorf("failed to add labelop: %w", err) 2243 + } 2244 + } 2245 + return tx.Commit() 2246 + } 2247 + 2248 + func (i *Ingester) deleteLabelOpRecord(ctx context.Context, did, rkey string, l *slog.Logger) error { 2249 + if err := i.materializeLabelOps(ctx, did, rkey, nil); err != nil { 2250 + return err 2251 + } 2252 + l.Info("ingested record") 2191 2253 return nil 2192 2254 }

+391

appview/ingester_label_test.go

··· 1 + package appview 2 + 3 + import ( 4 + "context" 5 + "encoding/json" 6 + "slices" 7 + "testing" 8 + "time" 9 + 10 + "github.com/bluesky-social/indigo/atproto/syntax" 11 + jmodels "github.com/bluesky-social/jetstream/pkg/models" 12 + "tangled.org/core/api/tangled" 13 + "tangled.org/core/appview/db" 14 + "tangled.org/core/appview/models" 15 + "tangled.org/core/orm" 16 + ) 17 + 18 + const ( 19 + labelOwner = "did:plc:boltless" 20 + labelRepoDid = "did:plc:anemone" 21 + labelPerform = "2026-06-01T00:00:00Z" 22 + ) 23 + 24 + func newLabelIngester(t *testing.T) *Ingester { 25 + t.Helper() 26 + ing := newStateIngester(t) 27 + ing.Acl = stubAcl{allow: true} 28 + return ing 29 + } 30 + 31 + func seedLabelDef(t *testing.T, d *db.DB, repoDid, defDid, defRkey, name string, scope []string, multiple, subscribe bool) syntax.ATURI { 32 + t.Helper() 33 + def := &models.LabelDefinition{ 34 + Did: defDid, Rkey: defRkey, Name: name, 35 + ValueType: models.ValueType{Type: models.ConcreteTypeString}, 36 + Scope: scope, Multiple: multiple, Created: time.Now(), 37 + } 38 + if _, err := db.AddLabelDefinition(d, def); err != nil { 39 + t.Fatalf("AddLabelDefinition: %v", err) 40 + } 41 + if subscribe { 42 + if err := db.SubscribeLabel(d, &models.RepoLabel{RepoDid: syntax.DID(repoDid), LabelAt: def.AtUri()}); err != nil { 43 + t.Fatalf("SubscribeLabel: %v", err) 44 + } 45 + } 46 + return def.AtUri() 47 + } 48 + 49 + func labelOpEvent(t *testing.T, op, did, rkey, subject, performedAt string, add, del [][2]string) *jmodels.Event { 50 + t.Helper() 51 + mk := func(pairs [][2]string) []*tangled.LabelOp_Operand { 52 + out := make([]*tangled.LabelOp_Operand, 0, len(pairs)) 53 + for _, p := range pairs { 54 + out = append(out, &tangled.LabelOp_Operand{Key: p[0], Value: p[1]}) 55 + } 56 + return out 57 + } 58 + raw, err := json.Marshal(tangled.LabelOp{Subject: subject, PerformedAt: performedAt, Add: mk(add), Delete: mk(del)}) 59 + if err != nil { 60 + t.Fatalf("marshal: %v", err) 61 + } 62 + return &jmodels.Event{Did: did, Kind: jmodels.EventKindCommit, Commit: &jmodels.Commit{ 63 + Operation: op, Collection: tangled.LabelOpNSID, RKey: rkey, Record: raw}} 64 + } 65 + 66 + func labelDefEvent(t *testing.T, did, rkey, name string, scope []string, multiple bool) *jmodels.Event { 67 + t.Helper() 68 + vt := tangled.LabelDefinition_ValueType{Type: string(models.ConcreteTypeString), Format: string(models.ValueTypeFormatAny)} 69 + raw, err := json.Marshal(tangled.LabelDefinition{Name: name, Scope: scope, Multiple: &multiple, CreatedAt: labelPerform, ValueType: &vt}) 70 + if err != nil { 71 + t.Fatalf("marshal: %v", err) 72 + } 73 + return &jmodels.Event{Did: did, Kind: jmodels.EventKindCommit, Commit: &jmodels.Commit{ 74 + Operation: jmodels.CommitOperationCreate, Collection: tangled.LabelDefinitionNSID, RKey: rkey, Record: raw}} 75 + } 76 + 77 + func mustIngestOp(t *testing.T, ing *Ingester, e *jmodels.Event) { 78 + t.Helper() 79 + if err := ing.ingestLabelOp(context.Background(), e, ing.Logger); err != nil { 80 + t.Fatalf("ingestLabelOp: %v", err) 81 + } 82 + } 83 + 84 + func issueAt(owner, rkey string) syntax.ATURI { 85 + return syntax.ATURI("at://" + owner + "/" + tangled.RepoIssueNSID + "/" + rkey) 86 + } 87 + 88 + func subjectLabels(t *testing.T, d *db.DB, subject syntax.ATURI) []string { 89 + t.Helper() 90 + states, err := db.GetLabels(d, orm.FilterEq("subject", subject)) 91 + if err != nil { 92 + t.Fatalf("GetLabels: %v", err) 93 + } 94 + vals := states[subject].LabelNameValues() 95 + slices.Sort(vals) 96 + return vals 97 + } 98 + 99 + func labelOpValues(t *testing.T, d *db.DB, did, rkey string) []string { 100 + t.Helper() 101 + ops, err := db.GetLabelOps(d, orm.FilterEq("did", did), orm.FilterEq("rkey", rkey)) 102 + if err != nil { 103 + t.Fatalf("GetLabelOps: %v", err) 104 + } 105 + out := make([]string, 0, len(ops)) 106 + for _, op := range ops { 107 + out = append(out, op.OperandValue) 108 + } 109 + slices.Sort(out) 110 + return out 111 + } 112 + 113 + func pendingCount(t *testing.T, d *db.DB, subject syntax.ATURI) int { 114 + t.Helper() 115 + p, err := db.PendingStateRecordsForSubject(d, subject) 116 + if err != nil { 117 + t.Fatalf("PendingStateRecordsForSubject: %v", err) 118 + } 119 + return len(p) 120 + } 121 + 122 + type labelStep struct { 123 + op, rkey, subj, perform string 124 + add, del []string 125 + } 126 + 127 + type foldCase struct { 128 + name string 129 + defRkey string 130 + defName string 131 + scope []string 132 + multiple bool 133 + unsub bool 134 + steps []labelStep 135 + want map[string][]string 136 + reversible bool 137 + } 138 + 139 + func runFold(t *testing.T, tc foldCase, steps []labelStep) map[string][]string { 140 + t.Helper() 141 + ing := newLabelIngester(t) 142 + defUri := "at://" + labelOwner + "/" + tangled.LabelDefinitionNSID + "/" + tc.defRkey 143 + 144 + seedRepo(t, ing.Db, labelOwner, labelRepoDid) 145 + seeded := map[string]bool{} 146 + for _, s := range steps { 147 + if !seeded[s.subj] { 148 + seedIssue(t, ing.Db, labelOwner, labelRepoDid, s.subj) 149 + seeded[s.subj] = true 150 + } 151 + } 152 + seedLabelDef(t, ing.Db, labelRepoDid, labelOwner, tc.defRkey, tc.defName, tc.scope, tc.multiple, !tc.unsub) 153 + 154 + pairs := func(vals []string) [][2]string { 155 + out := make([][2]string, 0, len(vals)) 156 + for _, v := range vals { 157 + out = append(out, [2]string{defUri, v}) 158 + } 159 + return out 160 + } 161 + kind := map[string]string{"create": jmodels.CommitOperationCreate, "update": jmodels.CommitOperationUpdate, "delete": jmodels.CommitOperationDelete} 162 + for _, s := range steps { 163 + perform := s.perform 164 + if perform == "" { 165 + perform = labelPerform 166 + } 167 + mustIngestOp(t, ing, labelOpEvent(t, kind[s.op], labelOwner, s.rkey, string(issueAt(labelOwner, s.subj)), perform, pairs(s.add), pairs(s.del))) 168 + } 169 + 170 + got := map[string][]string{} 171 + for subj := range tc.want { 172 + at := issueAt(labelOwner, subj) 173 + got[subj] = subjectLabels(t, ing.Db, at) 174 + if n := pendingCount(t, ing.Db, at); n != 0 { 175 + t.Fatalf("fold case %q must not park %s, pending=%d", tc.name, subj, n) 176 + } 177 + } 178 + return got 179 + } 180 + 181 + func TestIngestLabelOp_Fold(t *testing.T) { 182 + issue := []string{tangled.RepoIssueNSID} 183 + pull := []string{tangled.RepoPullNSID} 184 + 185 + tests := []foldCase{{ 186 + name: "delete removes the label", defRkey: "prio", defName: "priority", scope: issue, multiple: true, 187 + steps: []labelStep{{op: "create", rkey: "op1", subj: "issue1", add: []string{"high"}}, {op: "delete", rkey: "op1", subj: "issue1"}}, 188 + want: map[string][]string{"issue1": nil}, 189 + }, { 190 + name: "update drops operands the new record no longer carries", defRkey: "prio", defName: "priority", scope: issue, multiple: true, 191 + steps: []labelStep{{op: "create", rkey: "op1", subj: "issue1", add: []string{"high", "low"}}, {op: "update", rkey: "op1", subj: "issue1", add: []string{"high"}}}, 192 + want: map[string][]string{"issue1": {"priority:high"}}, 193 + }, { 194 + name: "update moving subject leaves no stale label", defRkey: "prio", defName: "priority", scope: issue, multiple: true, 195 + steps: []labelStep{{op: "create", rkey: "op1", subj: "issue1", add: []string{"high"}}, {op: "update", rkey: "op1", subj: "issue2", add: []string{"high"}}}, 196 + want: map[string][]string{"issue1": nil, "issue2": {"priority:high"}}, 197 + }, { 198 + name: "a pull-scoped label on an issue is dropped by the fold", defRkey: "prio", defName: "priority", scope: pull, multiple: true, 199 + steps: []labelStep{{op: "create", rkey: "op1", subj: "issue1", add: []string{"high"}}}, 200 + want: map[string][]string{"issue1": nil}, reversible: true, 201 + }, { 202 + name: "a globally-defined but unsubscribed def still applies", defRkey: "prio", defName: "priority", scope: issue, multiple: true, unsub: true, 203 + steps: []labelStep{{op: "create", rkey: "op1", subj: "issue1", add: []string{"high"}}}, 204 + want: map[string][]string{"issue1": {"priority:high"}}, reversible: true, 205 + }, { 206 + name: "a same-instant tie resolves by source rkey", defRkey: "st", defName: "status", scope: issue, multiple: false, 207 + steps: []labelStep{{op: "create", rkey: "aaa", subj: "issue1", add: []string{"open"}}, {op: "create", rkey: "bbb", subj: "issue1", add: []string{"closed"}}}, 208 + want: map[string][]string{"issue1": {"status:closed"}}, reversible: true, 209 + }} 210 + 211 + for _, tc := range tests { 212 + t.Run(tc.name, func(t *testing.T) { 213 + orders := [][]labelStep{tc.steps} 214 + if tc.reversible { 215 + rev := slices.Clone(tc.steps) 216 + slices.Reverse(rev) 217 + orders = append(orders, rev) 218 + } 219 + for _, steps := range orders { 220 + got := runFold(t, tc, steps) 221 + for subj, want := range tc.want { 222 + if !slices.Equal(got[subj], want) { 223 + t.Fatalf("subject %s: got %v want %v", subj, got[subj], want) 224 + } 225 + } 226 + } 227 + }) 228 + } 229 + } 230 + 231 + func TestIngestLabelOp_ReingestIsIdempotent(t *testing.T) { 232 + ing := newLabelIngester(t) 233 + subject := seedRepoAndIssue(t, ing.Db, labelOwner, labelRepoDid, "issue1") 234 + def := seedLabelDef(t, ing.Db, labelRepoDid, labelOwner, "prio", "priority", []string{tangled.RepoIssueNSID}, true, true) 235 + e := labelOpEvent(t, jmodels.CommitOperationCreate, labelOwner, "op1", string(subject), labelPerform, 236 + [][2]string{{def.String(), "high"}, {def.String(), "low"}}, nil) 237 + 238 + mustIngestOp(t, ing, e) 239 + first := labelOpValues(t, ing.Db, labelOwner, "op1") 240 + mustIngestOp(t, ing, e) 241 + if second := labelOpValues(t, ing.Db, labelOwner, "op1"); !slices.Equal(first, second) { 242 + t.Fatalf("re-ingesting an unchanged record changed its stored ops: %v -> %v", first, second) 243 + } 244 + if got := subjectLabels(t, ing.Db, subject); !slices.Equal(got, []string{"priority:high", "priority:low"}) { 245 + t.Fatalf("re-ingest changed the derived set: %v", got) 246 + } 247 + } 248 + 249 + func TestIngestLabelOp_Parking(t *testing.T) { 250 + defUri := "at://" + labelOwner + "/" + tangled.LabelDefinitionNSID + "/prio" 251 + issue := []string{tangled.RepoIssueNSID} 252 + seedDef := func(ing *Ingester) syntax.ATURI { 253 + return seedLabelDef(t, ing.Db, labelRepoDid, labelOwner, "prio", "priority", issue, true, true) 254 + } 255 + 256 + tests := []struct { 257 + name string 258 + missing string 259 + acl bool 260 + author string 261 + trigger string 262 + wantLabels []string 263 + }{ 264 + {"missing subject parks, drains on reconcile", "subject", true, labelOwner, "seed-subject", []string{"priority:high"}}, 265 + {"missing def parks, drains on reconcile", "def", true, labelOwner, "seed-def", []string{"priority:high"}}, 266 + {"def create drains a parked op without the sweep", "def", true, labelOwner, "def-event", []string{"priority:high"}}, 267 + {"deleting a parked op unparks it and a later def does not resurrect it", "def", true, labelOwner, "delete-then-def", nil}, 268 + {"a parked op that fails auth on drain is dropped", "subject", false, "did:plc:squid", "seed-subject", nil}, 269 + {"a missing-def park is evicted by TTL though its subject exists", "def", true, labelOwner, "evict", nil}, 270 + } 271 + 272 + for _, tt := range tests { 273 + t.Run(tt.name, func(t *testing.T) { 274 + ing := newLabelIngester(t) 275 + if !tt.acl { 276 + ing.Acl = stubAcl{} 277 + } 278 + subject := issueAt(labelOwner, "issue1") 279 + seedRepo(t, ing.Db, labelOwner, labelRepoDid) 280 + if tt.missing != "subject" { 281 + seedIssue(t, ing.Db, labelOwner, labelRepoDid, "issue1") 282 + } 283 + if tt.missing != "def" { 284 + seedDef(ing) 285 + } 286 + 287 + mustIngestOp(t, ing, labelOpEvent(t, jmodels.CommitOperationCreate, tt.author, "op1", string(subject), labelPerform, 288 + [][2]string{{defUri, "high"}}, nil)) 289 + if n := pendingCount(t, ing.Db, subject); n != 1 { 290 + t.Fatalf("op must park while its %s is missing, pending=%d", tt.missing, n) 291 + } 292 + 293 + switch tt.trigger { 294 + case "seed-subject": 295 + seedIssue(t, ing.Db, labelOwner, labelRepoDid, "issue1") 296 + ing.ReconcilePendingState() 297 + case "seed-def": 298 + seedDef(ing) 299 + ing.ReconcilePendingState() 300 + case "def-event": 301 + if err := ing.ingestLabelDefinition(labelDefEvent(t, labelOwner, "prio", "priority", issue, true), ing.Logger); err != nil { 302 + t.Fatalf("ingest def: %v", err) 303 + } 304 + case "delete-then-def": 305 + mustIngestOp(t, ing, labelOpEvent(t, jmodels.CommitOperationDelete, tt.author, "op1", string(subject), "", nil, nil)) 306 + if n := pendingCount(t, ing.Db, subject); n != 0 { 307 + t.Fatalf("deleting a parked op must unpark it, pending=%d", n) 308 + } 309 + seedDef(ing) 310 + ing.ReconcilePendingState() 311 + case "evict": 312 + if n, err := db.EvictStalePendingStateRecords(ing.Db, "2999-01-01T00:00:00Z"); err != nil { 313 + t.Fatalf("evict: %v", err) 314 + } else if n != 1 { 315 + t.Fatalf("a stale missing-def park must be evicted, got %d", n) 316 + } 317 + } 318 + 319 + if got := subjectLabels(t, ing.Db, subject); !slices.Equal(got, tt.wantLabels) { 320 + t.Fatalf("after %q: got %v want %v", tt.trigger, got, tt.wantLabels) 321 + } 322 + if n := pendingCount(t, ing.Db, subject); n != 0 { 323 + t.Fatalf("no record must remain parked, pending=%d", n) 324 + } 325 + }) 326 + } 327 + } 328 + 329 + func TestIngestLabelOp_DeauthorizedUpdatePreservesPriorOps(t *testing.T) { 330 + ing := newLabelIngester(t) 331 + author := "did:plc:squid" 332 + subject := seedRepoAndIssue(t, ing.Db, labelOwner, labelRepoDid, "issue1") 333 + def := seedLabelDef(t, ing.Db, labelRepoDid, labelOwner, "prio", "priority", []string{tangled.RepoIssueNSID}, true, true) 334 + 335 + mustIngestOp(t, ing, labelOpEvent(t, jmodels.CommitOperationCreate, author, "op1", string(subject), labelPerform, 336 + [][2]string{{def.String(), "high"}}, nil)) 337 + if got := subjectLabels(t, ing.Db, subject); !slices.Equal(got, []string{"priority:high"}) { 338 + t.Fatalf("authorized create: got %v", got) 339 + } 340 + 341 + ing.Acl = stubAcl{} 342 + mustIngestOp(t, ing, labelOpEvent(t, jmodels.CommitOperationUpdate, author, "op1", string(subject), labelPerform, 343 + [][2]string{{def.String(), "low"}}, nil)) 344 + if got := subjectLabels(t, ing.Db, subject); !slices.Equal(got, []string{"priority:high"}) { 345 + t.Fatalf("a now-unauthorized update must neither apply nor erase the prior authorized op, got %v", got) 346 + } 347 + } 348 + 349 + func TestIngestLabelOp_DeauthorizedRedeliveryPreservesPriorOps(t *testing.T) { 350 + ing := newLabelIngester(t) 351 + author := "did:plc:squid" 352 + subject := seedRepoAndIssue(t, ing.Db, labelOwner, labelRepoDid, "issue1") 353 + def := seedLabelDef(t, ing.Db, labelRepoDid, labelOwner, "prio", "priority", []string{tangled.RepoIssueNSID}, true, true) 354 + 355 + create := labelOpEvent(t, jmodels.CommitOperationCreate, author, "op1", string(subject), labelPerform, 356 + [][2]string{{def.String(), "high"}}, nil) 357 + mustIngestOp(t, ing, create) 358 + if got := subjectLabels(t, ing.Db, subject); !slices.Equal(got, []string{"priority:high"}) { 359 + t.Fatalf("authorized create: got %v", got) 360 + } 361 + 362 + ing.Acl = stubAcl{} 363 + mustIngestOp(t, ing, create) 364 + if got := subjectLabels(t, ing.Db, subject); !slices.Equal(got, []string{"priority:high"}) { 365 + t.Fatalf("a redelivered create must not erase a label applied while authorized, got %v", got) 366 + } 367 + } 368 + 369 + func TestIngestLabelOp_DefDeleteDropsOps(t *testing.T) { 370 + ing := newLabelIngester(t) 371 + subject := seedRepoAndIssue(t, ing.Db, labelOwner, labelRepoDid, "issue1") 372 + def := seedLabelDef(t, ing.Db, labelRepoDid, labelOwner, "prio", "priority", []string{tangled.RepoIssueNSID}, true, true) 373 + 374 + mustIngestOp(t, ing, labelOpEvent(t, jmodels.CommitOperationCreate, labelOwner, "op1", string(subject), labelPerform, 375 + [][2]string{{def.String(), "high"}}, nil)) 376 + if got := subjectLabels(t, ing.Db, subject); !slices.Equal(got, []string{"priority:high"}) { 377 + t.Fatalf("after create: got %v", got) 378 + } 379 + 380 + del := &jmodels.Event{Did: labelOwner, Kind: jmodels.EventKindCommit, Commit: &jmodels.Commit{ 381 + Operation: jmodels.CommitOperationDelete, Collection: tangled.LabelDefinitionNSID, RKey: "prio"}} 382 + if err := ing.ingestLabelDefinition(del, ing.Logger); err != nil { 383 + t.Fatalf("ingest def delete: %v", err) 384 + } 385 + if rows := labelOpValues(t, ing.Db, labelOwner, "op1"); len(rows) != 0 { 386 + t.Fatalf("deleting a def must cascade-delete its label_ops rows, survived: %v", rows) 387 + } 388 + if got := subjectLabels(t, ing.Db, subject); len(got) != 0 { 389 + t.Fatalf("deleting a def must drop its labels, got %v", got) 390 + } 391 + }

+5 -1

appview/ingester_repo.go

··· 250 250 if err := applyRepoMetadata(tx, current, desired); err != nil { 251 251 return fmt.Errorf("failed to apply repo metadata: %w", err) 252 252 } 253 - return tx.Commit() 253 + if err := tx.Commit(); err != nil { 254 + return err 255 + } 256 + 257 + return nil 254 258 } 255 259 256 260 func (i *Ingester) ingestRepoDelete(ctx context.Context, e *jmodels.Event, l *slog.Logger) error {

+36 -17

appview/ingester_state_test.go

··· 2 2 3 3 import ( 4 4 "context" 5 + "database/sql" 5 6 "encoding/json" 6 7 "errors" 7 8 "log/slog" ··· 41 42 return s.allow, s.err 42 43 } 43 44 44 - func seedRepoAndIssue(t *testing.T, d *db.DB, ownerDid, repoDid, issueRkey string) syntax.ATURI { 45 + func seedTx(t *testing.T, d *db.DB, fn func(tx *sql.Tx) error) { 45 46 t.Helper() 46 47 tx, err := d.Begin() 47 48 if err != nil { 48 49 t.Fatalf("Begin: %v", err) 49 50 } 50 - if err := db.AddRepo(tx, &models.Repo{ 51 - Did: ownerDid, 52 - Name: "anemone", 53 - Knot: "knot.example", 54 - Rkey: "anemone", 55 - RepoDid: repoDid, 56 - }); err != nil { 57 - t.Fatalf("AddRepo: %v", err) 51 + defer tx.Rollback() 52 + if err := fn(tx); err != nil { 53 + t.Fatalf("seed: %v", err) 58 54 } 55 + if err := tx.Commit(); err != nil { 56 + t.Fatalf("Commit: %v", err) 57 + } 58 + } 59 + 60 + func seedRepo(t *testing.T, d *db.DB, ownerDid, repoDid string) { 61 + t.Helper() 62 + seedTx(t, d, func(tx *sql.Tx) error { 63 + return db.AddRepo(tx, &models.Repo{ 64 + Did: ownerDid, 65 + Name: "anemone", 66 + Knot: "knot.example", 67 + Rkey: "anemone", 68 + RepoDid: repoDid, 69 + }) 70 + }) 71 + } 72 + 73 + func seedIssue(t *testing.T, d *db.DB, ownerDid, repoDid, issueRkey string) syntax.ATURI { 74 + t.Helper() 59 75 issue := &models.Issue{ 60 76 Did: ownerDid, 61 77 Rkey: issueRkey, ··· 64 80 Body: "body", 65 81 Open: true, 66 82 } 67 - if err := db.PutIssue(tx, issue); err != nil { 68 - t.Fatalf("PutIssue: %v", err) 69 - } 70 - if err := tx.Commit(); err != nil { 71 - t.Fatalf("Commit: %v", err) 72 - } 83 + seedTx(t, d, func(tx *sql.Tx) error { 84 + return db.PutIssue(tx, issue) 85 + }) 73 86 return issue.AtUri() 87 + } 88 + 89 + func seedRepoAndIssue(t *testing.T, d *db.DB, ownerDid, repoDid, issueRkey string) syntax.ATURI { 90 + t.Helper() 91 + seedRepo(t, d, ownerDid, repoDid) 92 + return seedIssue(t, d, ownerDid, repoDid, issueRkey) 74 93 } 75 94 76 95 func issueStateEvent(t *testing.T, op, did, rkey, subject, state, createdAt string) *jmodels.Event { ··· 194 213 } 195 214 196 215 at := seedRepoAndIssue(t, ing.Db, owner, "did:plc:anemone", "issue1") 197 - ing.drainPendingState(ctx, at, issueStateSpec, ing.Logger) 216 + ing.drainPendingState(ctx, at, ing.Logger) 198 217 if !ingestedIssueOpen(t, ing.Db, at) { 199 218 t.Fatal("a deleted parked record must not apply after the subject arrives") 200 219 } ··· 216 235 } 217 236 218 237 at := seedRepoAndIssue(t, ing.Db, owner, "did:plc:anemone", "issue1") 219 - ing.drainPendingState(ctx, at, issueStateSpec, ing.Logger) 238 + ing.drainPendingState(ctx, at, ing.Logger) 220 239 221 240 if !ingestedIssueOpen(t, ing.Db, at) { 222 241 t.Fatal("a parked record that fails authorization on drain must not apply")

Configure Feed

Configure Feed