Monorepo for Tangled
appview/ingester: ingest issue/pull state
Lewis: May this revision serve well! <lewis@tangled.org>
+304
-1
+304
-1
appview/ingester.go
+304
-1
appview/ingester.go
···
33
33
"tangled.org/core/appview/notify"
34
34
"tangled.org/core/appview/repoverify"
35
35
"tangled.org/core/appview/serververify"
36
+
"tangled.org/core/consts"
36
37
"tangled.org/core/idresolver"
37
38
"tangled.org/core/orm"
38
39
"tangled.org/core/rbac"
39
40
)
40
41
42
+
type RepoPermissionChecker interface {
43
+
HasRepoPermissionErr(ctx context.Context, repo *models.Repo, userDid, perm string) (bool, error)
44
+
}
45
+
41
46
type Ingester struct {
42
47
Ctx context.Context
43
48
Db *db.DB
44
49
Enforcer *rbac.Enforcer
45
-
Acl *knotacl.Service
50
+
Acl RepoPermissionChecker
46
51
IdResolver *idresolver.Resolver
47
52
Cache *cache.Cache
48
53
Config *config.Config
···
105
110
err = i.ingestString(e, l)
106
111
case tangled.RepoIssueNSID:
107
112
err = i.ingestIssue(ctx, e, l)
113
+
case tangled.RepoIssueStateNSID:
114
+
err = i.ingestState(ctx, e, l, issueStateSpec)
108
115
case tangled.RepoPullNSID:
109
116
err = i.ingestPull(ctx, e, l)
117
+
case tangled.RepoPullStatusNSID:
118
+
err = i.ingestState(ctx, e, l, pullStatusSpec)
110
119
case tangled.FeedCommentNSID:
111
120
err = i.ingestComment(e, l)
112
121
case tangled.RepoIssueCommentNSID:
···
1393
1402
return err
1394
1403
}
1395
1404
1405
+
if err := db.ResolveIssueState(tx, issue.AtUri()); err != nil {
1406
+
l.Error("failed to resolve issue state", "err", err)
1407
+
return err
1408
+
}
1409
+
1396
1410
err = tx.Commit()
1397
1411
if err != nil {
1398
1412
l.Error("failed to commit txn", "err", err)
1399
1413
return err
1400
1414
}
1415
+
1416
+
i.drainPendingState(ctx, issue.AtUri(), issueStateSpec, l)
1401
1417
1402
1418
l.Info("ingested record")
1403
1419
return nil
···
1553
1569
return err
1554
1570
}
1555
1571
1572
+
if err := db.ResolvePullStatus(tx, pull.AtUri()); err != nil {
1573
+
l.Error("failed to resolve pull status", "err", err)
1574
+
return err
1575
+
}
1576
+
1556
1577
err = tx.Commit()
1557
1578
if err != nil {
1558
1579
l.Error("failed to commit txn", "err", err)
1559
1580
return err
1560
1581
}
1582
+
1583
+
i.drainPendingState(ctx, pull.AtUri(), pullStatusSpec, l)
1561
1584
1562
1585
l.Info("ingested record")
1563
1586
return nil
···
1588
1611
}
1589
1612
1590
1613
return nil
1614
+
}
1615
+
1616
+
func (i *Ingester) authorizeStateRecord(ctx context.Context, repo *models.Repo, subjectAuthorDid, recordAuthorDid string, l *slog.Logger) (bool, error) {
1617
+
if recordAuthorDid == subjectAuthorDid {
1618
+
return true, nil
1619
+
}
1620
+
if recordAuthorDid == consts.TangledDid {
1621
+
return true, nil
1622
+
}
1623
+
1624
+
ok, err := i.Acl.HasRepoPermissionErr(ctx, repo, recordAuthorDid, "repo:push")
1625
+
if err != nil {
1626
+
if errors.Is(err, knotacl.ErrKnotUnreachable) {
1627
+
l.Warn("ingesting state record without permission check", "did", recordAuthorDid, "err", err)
1628
+
return true, nil
1629
+
}
1630
+
return false, err
1631
+
}
1632
+
return ok, nil
1633
+
}
1634
+
1635
+
type stateIngestSpec struct {
1636
+
subjectNSID string
1637
+
parse func(did, rkey string, raw json.RawMessage) (models.StateRecord, error)
1638
+
findSubject func(e db.Execer, subject syntax.ATURI) (repo *models.Repo, authorDid string, found bool, err error)
1639
+
put func(tx *sql.Tx, rec models.StateRecord) (syntax.ATURI, error)
1640
+
resolve func(tx *sql.Tx, subject syntax.ATURI) error
1641
+
recompute func(tx *sql.Tx, subject syntax.ATURI) error
1642
+
del func(tx *sql.Tx, did, rkey string) (syntax.ATURI, error)
1643
+
}
1644
+
1645
+
var issueStateSpec = stateIngestSpec{
1646
+
subjectNSID: tangled.RepoIssueNSID,
1647
+
parse: func(did, rkey string, raw json.RawMessage) (models.StateRecord, error) {
1648
+
record := tangled.RepoIssueState{}
1649
+
if err := json.Unmarshal(raw, &record); err != nil {
1650
+
return models.StateRecord{}, fmt.Errorf("invalid record: %w", err)
1651
+
}
1652
+
return models.IssueStateFromRecord(did, rkey, record)
1653
+
},
1654
+
findSubject: func(e db.Execer, subject syntax.ATURI) (*models.Repo, string, bool, error) {
1655
+
issues, err := db.GetIssues(e, orm.FilterEq("at_uri", subject))
1656
+
if err != nil {
1657
+
return nil, "", false, err
1658
+
}
1659
+
if len(issues) != 1 || issues[0].Repo == nil {
1660
+
return nil, "", false, nil
1661
+
}
1662
+
return issues[0].Repo, issues[0].Did, true, nil
1663
+
},
1664
+
put: db.PutIssueState,
1665
+
resolve: db.ResolveIssueState,
1666
+
recompute: db.RecomputeIssueState,
1667
+
del: db.DeleteIssueState,
1668
+
}
1669
+
1670
+
var pullStatusSpec = stateIngestSpec{
1671
+
subjectNSID: tangled.RepoPullNSID,
1672
+
parse: func(did, rkey string, raw json.RawMessage) (models.StateRecord, error) {
1673
+
record := tangled.RepoPullStatus{}
1674
+
if err := json.Unmarshal(raw, &record); err != nil {
1675
+
return models.StateRecord{}, fmt.Errorf("invalid record: %w", err)
1676
+
}
1677
+
return models.PullStatusFromRecord(did, rkey, record)
1678
+
},
1679
+
findSubject: func(e db.Execer, subject syntax.ATURI) (*models.Repo, string, bool, error) {
1680
+
pulls, err := db.GetPulls(e, orm.FilterEq("at_uri", subject))
1681
+
if err != nil {
1682
+
return nil, "", false, err
1683
+
}
1684
+
if len(pulls) != 1 || pulls[0].Repo == nil {
1685
+
return nil, "", false, nil
1686
+
}
1687
+
return pulls[0].Repo, pulls[0].OwnerDid, true, nil
1688
+
},
1689
+
put: db.PutPullStatus,
1690
+
resolve: db.ResolvePullStatus,
1691
+
recompute: db.RecomputePullStatus,
1692
+
del: db.DeletePullStatus,
1693
+
}
1694
+
1695
+
func (i *Ingester) ingestState(ctx context.Context, e *jmodels.Event, l *slog.Logger, spec stateIngestSpec) error {
1696
+
did := e.Did
1697
+
rkey := e.Commit.RKey
1698
+
nsid := e.Commit.Collection
1699
+
1700
+
l = l.With("handler", "ingestState", "nsid", nsid)
1701
+
1702
+
switch e.Commit.Operation {
1703
+
case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate:
1704
+
return i.applyStateRecord(ctx, did, rkey, nsid, e.Commit.Record, spec, l)
1705
+
case jmodels.CommitOperationDelete:
1706
+
return i.deleteStateRecord(ctx, did, rkey, nsid, spec, l)
1707
+
}
1708
+
1709
+
return nil
1710
+
}
1711
+
1712
+
func (i *Ingester) applyStateRecord(ctx context.Context, did, rkey, nsid string, raw []byte, spec stateIngestSpec, l *slog.Logger) error {
1713
+
rec, err := spec.parse(did, rkey, json.RawMessage(raw))
1714
+
if err != nil {
1715
+
return err
1716
+
}
1717
+
if string(rec.Subject.Collection()) != spec.subjectNSID {
1718
+
return fmt.Errorf("state subject is not %s: %s", spec.subjectNSID, rec.Subject)
1719
+
}
1720
+
1721
+
repo, authorDid, found, err := spec.findSubject(i.Db, rec.Subject)
1722
+
if err != nil {
1723
+
return fmt.Errorf("failed to look up state subject: %w", err)
1724
+
}
1725
+
if !found {
1726
+
return i.parkStateRecord(ctx, did, rkey, nsid, rec.Subject, raw, l)
1727
+
}
1728
+
1729
+
authorized, err := i.authorizeStateRecord(ctx, repo, authorDid, did, l)
1730
+
if err != nil {
1731
+
return i.parkStateRecord(ctx, did, rkey, nsid, rec.Subject, raw, l)
1732
+
}
1733
+
1734
+
tx, err := i.Db.BeginTx(ctx, nil)
1735
+
if err != nil {
1736
+
return err
1737
+
}
1738
+
defer tx.Rollback()
1739
+
1740
+
if err := db.UnparkStateRecord(tx, did, rkey, nsid); err != nil {
1741
+
return fmt.Errorf("failed to unpark state record: %w", err)
1742
+
}
1743
+
1744
+
if !authorized {
1745
+
if err := tx.Commit(); err != nil {
1746
+
return err
1747
+
}
1748
+
l.Warn("dropped unauthorized state record", "did", did, "rkey", rkey, "subject", rec.Subject)
1749
+
return nil
1750
+
}
1751
+
1752
+
priorSubject, err := spec.put(tx, rec)
1753
+
if err != nil {
1754
+
return fmt.Errorf("failed to put state record: %w", err)
1755
+
}
1756
+
if err := spec.resolve(tx, rec.Subject); err != nil {
1757
+
return fmt.Errorf("failed to resolve state: %w", err)
1758
+
}
1759
+
if priorSubject != "" {
1760
+
if err := spec.recompute(tx, priorSubject); err != nil {
1761
+
return fmt.Errorf("failed to recompute prior subject state: %w", err)
1762
+
}
1763
+
}
1764
+
1765
+
if err := tx.Commit(); err != nil {
1766
+
return err
1767
+
}
1768
+
1769
+
l.Info("ingested record")
1770
+
return nil
1771
+
}
1772
+
1773
+
func (i *Ingester) deleteStateRecord(ctx context.Context, did, rkey, nsid string, spec stateIngestSpec, l *slog.Logger) error {
1774
+
tx, err := i.Db.BeginTx(ctx, nil)
1775
+
if err != nil {
1776
+
return err
1777
+
}
1778
+
defer tx.Rollback()
1779
+
1780
+
if err := db.UnparkStateRecord(tx, did, rkey, nsid); err != nil {
1781
+
return fmt.Errorf("failed to unpark state record: %w", err)
1782
+
}
1783
+
subject, err := spec.del(tx, did, rkey)
1784
+
if err != nil {
1785
+
return fmt.Errorf("failed to delete state record: %w", err)
1786
+
}
1787
+
if subject != "" {
1788
+
if err := spec.recompute(tx, subject); err != nil {
1789
+
return fmt.Errorf("failed to recompute state: %w", err)
1790
+
}
1791
+
}
1792
+
1793
+
if err := tx.Commit(); err != nil {
1794
+
return err
1795
+
}
1796
+
1797
+
l.Info("ingested record")
1798
+
return nil
1799
+
}
1800
+
1801
+
func (i *Ingester) parkStateRecord(ctx context.Context, did, rkey, nsid string, subject syntax.ATURI, raw []byte, l *slog.Logger) error {
1802
+
tx, err := i.Db.BeginTx(ctx, nil)
1803
+
if err != nil {
1804
+
return err
1805
+
}
1806
+
defer tx.Rollback()
1807
+
1808
+
if err := db.ParkStateRecord(tx, db.PendingStateRecord{
1809
+
Did: did,
1810
+
Rkey: rkey,
1811
+
Nsid: nsid,
1812
+
Subject: subject,
1813
+
Record: raw,
1814
+
}); err != nil {
1815
+
return fmt.Errorf("failed to park state record: %w", err)
1816
+
}
1817
+
1818
+
if err := tx.Commit(); err != nil {
1819
+
return err
1820
+
}
1821
+
1822
+
l.Info("parked state record for retry", "subject", subject)
1823
+
return nil
1824
+
}
1825
+
1826
+
func (i *Ingester) drainPendingState(ctx context.Context, subject syntax.ATURI, spec stateIngestSpec, l *slog.Logger) {
1827
+
pending, err := db.PendingStateRecordsForSubject(i.Db, subject)
1828
+
if err != nil {
1829
+
l.Error("failed to load pending state records", "err", err, "subject", subject)
1830
+
return
1831
+
}
1832
+
for _, p := range pending {
1833
+
if err := i.applyStateRecord(ctx, p.Did, p.Rkey, p.Nsid, p.Record, spec, l); err != nil {
1834
+
l.Error("failed to drain pending state record", "err", err, "did", p.Did, "rkey", p.Rkey)
1835
+
}
1836
+
}
1837
+
}
1838
+
1839
+
const (
1840
+
pendingStateReconcileInterval = time.Hour
1841
+
pendingStateRecordTTL = 7 * 24 * time.Hour
1842
+
)
1843
+
1844
+
func stateSpecForSubject(subject syntax.ATURI) (stateIngestSpec, bool) {
1845
+
switch string(subject.Collection()) {
1846
+
case tangled.RepoIssueNSID:
1847
+
return issueStateSpec, true
1848
+
case tangled.RepoPullNSID:
1849
+
return pullStatusSpec, true
1850
+
default:
1851
+
return stateIngestSpec{}, false
1852
+
}
1853
+
}
1854
+
1855
+
func (i *Ingester) StartPendingStateReconciler() {
1856
+
i.ReconcilePendingState()
1857
+
1858
+
ticker := time.NewTicker(pendingStateReconcileInterval)
1859
+
defer ticker.Stop()
1860
+
for {
1861
+
select {
1862
+
case <-i.Ctx.Done():
1863
+
return
1864
+
case <-ticker.C:
1865
+
i.ReconcilePendingState()
1866
+
}
1867
+
}
1868
+
}
1869
+
1870
+
func (i *Ingester) ReconcilePendingState() {
1871
+
l := i.Logger.With("handler", "reconcilePendingState")
1872
+
1873
+
subjects, err := db.DistinctPendingStateSubjects(i.Db)
1874
+
if err != nil {
1875
+
l.Error("failed to list pending state subjects", "err", err)
1876
+
}
1877
+
for _, subject := range subjects {
1878
+
spec, ok := stateSpecForSubject(subject)
1879
+
if !ok {
1880
+
continue
1881
+
}
1882
+
i.drainPendingState(i.Ctx, subject, spec, l)
1883
+
}
1884
+
1885
+
cutoff := time.Now().Add(-pendingStateRecordTTL).UTC().Format(time.RFC3339)
1886
+
evicted, err := db.EvictStalePendingStateRecords(i.Db, cutoff)
1887
+
if err != nil {
1888
+
l.Error("failed to evict stale pending state records", "err", err)
1889
+
return
1890
+
}
1891
+
if evicted > 0 {
1892
+
l.Warn("evicted stale pending state records", "count", evicted, "olderThan", cutoff)
1893
+
}
1591
1894
}
1592
1895
1593
1896
// ingestIssueComment ingests legacy sh.tangled.repo.issue.comment deletions