+107 -96

appview/ingester.go

Reviewed

··· 72 72 case jmodels.EventKindIdentity: 73 73 err = i.IdResolver.InvalidateIdent(ctx, e.Identity.Did) 74 74 case jmodels.EventKindCommit: 75 75 + l = l.With( 76 76 + "nsid", e.Commit.Collection, 77 77 + "did", e.Did, 78 78 + "rkey", e.Commit.RKey, 79 79 + "op", e.Commit.Operation, 80 80 + ) 75 81 switch e.Commit.Collection { 76 82 case tangled.GraphFollowNSID: 77 77 - err = i.ingestFollow(e) 83 83 + err = i.ingestFollow(e, l) 78 84 case tangled.GraphVouchNSID: 79 79 - err = i.ingestVouch(ctx, e) 85 85 + err = i.ingestVouch(ctx, e, l) 80 86 case tangled.FeedStarNSID: 81 81 - err = i.ingestStar(ctx, e) 87 87 + err = i.ingestStar(ctx, e, l) 82 88 case tangled.FeedReactionNSID: 83 83 - err = i.ingestReaction(e) 89 89 + err = i.ingestReaction(e, l) 84 90 case tangled.PublicKeyNSID: 85 85 - err = i.ingestPublicKey(e) 91 91 + err = i.ingestPublicKey(e, l) 86 92 case tangled.RepoArtifactNSID: 87 87 - err = i.ingestArtifact(ctx, e) 93 93 + err = i.ingestArtifact(ctx, e, l) 88 94 case tangled.ActorProfileNSID: 89 89 - err = i.ingestProfile(ctx, e) 95 95 + err = i.ingestProfile(ctx, e, l) 90 96 case tangled.SpindleMemberNSID: 91 91 - err = i.ingestSpindleMember(ctx, e) 97 97 + err = i.ingestSpindleMember(ctx, e, l) 92 98 case tangled.SpindleNSID: 93 93 - err = i.ingestSpindle(ctx, e) 99 99 + err = i.ingestSpindle(ctx, e, l) 94 100 case tangled.KnotMemberNSID: 95 95 - err = i.ingestKnotMember(ctx, e) 101 101 + err = i.ingestKnotMember(ctx, e, l) 96 102 case tangled.KnotNSID: 97 97 - err = i.ingestKnot(ctx, e) 103 103 + err = i.ingestKnot(ctx, e, l) 98 104 case tangled.StringNSID: 99 99 - err = i.ingestString(e) 105 105 + err = i.ingestString(e, l) 100 106 case tangled.RepoIssueNSID: 101 101 - err = i.ingestIssue(ctx, e) 107 107 + err = i.ingestIssue(ctx, e, l) 102 108 case tangled.RepoPullNSID: 103 103 - err = i.ingestPull(ctx, e) 109 109 + err = i.ingestPull(ctx, e, l) 104 110 case tangled.FeedCommentNSID: 105 105 - err = i.ingestComment(e) 111 111 + err = i.ingestComment(e, l) 106 112 case tangled.RepoIssueCommentNSID: 107 107 - err = i.ingestIssueComment(e) 113 113 + err = i.ingestIssueComment(e, l) 108 114 case tangled.RepoPullCommentNSID: 109 109 - err = i.ingestPullComment(e) 115 115 + err = i.ingestPullComment(e, l) 110 116 case tangled.LabelDefinitionNSID: 111 111 - err = i.ingestLabelDefinition(e) 117 117 + err = i.ingestLabelDefinition(e, l) 112 118 case tangled.LabelOpNSID: 113 113 - err = i.ingestLabelOp(e) 119 119 + err = i.ingestLabelOp(e, l) 114 120 case tangled.RepoNSID: 115 115 - err = i.ingestRepo(ctx, e) 121 121 + err = i.ingestRepo(ctx, e, l) 116 122 } 117 117 - l = i.Logger.With("nsid", e.Commit.Collection) 118 123 } 119 124 120 125 if err != nil { ··· 183 188 } 184 189 } 185 190 186 186 - func (i *Ingester) ingestStar(ctx context.Context, e *jmodels.Event) error { 191 191 + func (i *Ingester) ingestStar(ctx context.Context, e *jmodels.Event, l *slog.Logger) error { 187 192 var err error 188 193 did := e.Did 189 194 190 190 - l := i.Logger.With("handler", "ingestStar") 191 191 - l = l.With("nsid", e.Commit.Collection) 195 195 + l = l.With("handler", "ingestStar") 192 196 193 197 switch e.Commit.Operation { 194 198 case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: ··· 241 245 return fmt.Errorf("failed to %s star record: %w", e.Commit.Operation, err) 242 246 } 243 247 248 248 + l.Info("ingested record") 244 249 return nil 245 250 } 246 251 247 247 - func (i *Ingester) ingestFollow(e *jmodels.Event) error { 252 252 + func (i *Ingester) ingestFollow(e *jmodels.Event, l *slog.Logger) error { 248 253 var err error 249 254 did := e.Did 250 255 251 251 - l := i.Logger.With("handler", "ingestFollow") 252 252 - l = l.With("nsid", e.Commit.Collection) 256 256 + l = l.With("handler", "ingestFollow") 253 257 254 258 switch e.Commit.Operation { 255 259 case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: ··· 274 278 return fmt.Errorf("failed to %s follow record: %w", e.Commit.Operation, err) 275 279 } 276 280 281 281 + l.Info("ingested record") 277 282 return nil 278 283 } 279 284 280 280 - func (i *Ingester) ingestVouch(ctx context.Context, e *jmodels.Event) error { 285 285 + func (i *Ingester) ingestVouch(ctx context.Context, e *jmodels.Event, l *slog.Logger) error { 281 286 var err error 282 287 did := e.Did 283 288 284 284 - l := i.Logger.With("handler", "ingestVouch") 285 285 - l = l.With("nsid", e.Commit.Collection) 286 286 - l.Info("ingesting vouch") 289 289 + l = l.With("handler", "ingestVouch") 287 290 288 291 switch e.Commit.Operation { 289 292 case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: ··· 368 371 return fmt.Errorf("failed to %s vouch record: %w", e.Commit.Operation, err) 369 372 } 370 373 374 374 + l.Info("ingested record") 371 375 return nil 372 376 } 373 377 374 374 - func (i *Ingester) ingestPublicKey(e *jmodels.Event) error { 378 378 + func (i *Ingester) ingestPublicKey(e *jmodels.Event, l *slog.Logger) error { 375 379 did := e.Did 376 380 var err error 377 381 378 378 - l := i.Logger.With("handler", "ingestPublicKey") 379 379 - l = l.With("nsid", e.Commit.Collection) 382 382 + l = l.With("handler", "ingestPublicKey") 380 383 381 384 switch e.Commit.Operation { 382 385 case jmodels.CommitOperationCreate: ··· 414 417 return fmt.Errorf("failed to %s pubkey record: %w", e.Commit.Operation, err) 415 418 } 416 419 420 420 + l.Info("ingested record") 417 421 return nil 418 422 } 419 423 420 420 - func (i *Ingester) ingestArtifact(ctx context.Context, e *jmodels.Event) error { 424 424 + func (i *Ingester) ingestArtifact(ctx context.Context, e *jmodels.Event, l *slog.Logger) error { 421 425 did := e.Did 422 426 var err error 423 427 424 424 - l := i.Logger.With("handler", "ingestArtifact") 425 425 - l = l.With("nsid", e.Commit.Collection) 428 428 + l = l.With("handler", "ingestArtifact") 426 429 427 430 switch e.Commit.Operation { 428 431 case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: ··· 496 499 return fmt.Errorf("failed to %s artifact record: %w", e.Commit.Operation, err) 497 500 } 498 501 502 502 + l.Info("ingested record") 499 503 return nil 500 504 } 501 505 502 502 - func (i *Ingester) ingestProfile(ctx context.Context, e *jmodels.Event) error { 506 506 + func (i *Ingester) ingestProfile(ctx context.Context, e *jmodels.Event, l *slog.Logger) error { 503 507 did := e.Did 504 508 var err error 505 509 506 506 - l := i.Logger.With("handler", "ingestProfile") 507 507 - l = l.With("nsid", e.Commit.Collection) 510 510 + l = l.With("handler", "ingestProfile") 508 511 509 512 if e.Commit.RKey != "self" { 510 513 return fmt.Errorf("ingestProfile only ingests `self` record") ··· 638 641 return fmt.Errorf("failed to %s profile record: %w", e.Commit.Operation, err) 639 642 } 640 643 644 644 + l.Info("ingested record") 641 645 return nil 642 646 } 643 647 644 644 - func (i *Ingester) ingestSpindleMember(ctx context.Context, e *jmodels.Event) error { 648 648 + func (i *Ingester) ingestSpindleMember(ctx context.Context, e *jmodels.Event, l *slog.Logger) error { 645 649 did := e.Did 646 650 var err error 647 651 648 648 - l := i.Logger.With("handler", "ingestSpindleMember") 649 649 - l = l.With("nsid", e.Commit.Collection) 652 652 + l = l.With("handler", "ingestSpindleMember") 650 653 651 654 switch e.Commit.Operation { 652 655 case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: ··· 804 807 return nil 805 808 } 806 809 807 807 - func (i *Ingester) ingestSpindle(ctx context.Context, e *jmodels.Event) error { 810 810 + func (i *Ingester) ingestSpindle(ctx context.Context, e *jmodels.Event, l *slog.Logger) error { 808 811 did := e.Did 809 812 var err error 810 813 811 811 - l := i.Logger.With("handler", "ingestSpindle") 812 812 - l = l.With("nsid", e.Commit.Collection) 814 814 + l = l.With("handler", "ingestSpindle") 813 815 814 816 switch e.Commit.Operation { 815 817 case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: ··· 836 838 l.Warn("failed to verify spindle", "instance", instance, "did", did, "err", err) 837 839 } 838 840 841 841 + l.Info("ingested record", "instance", instance) 839 842 return nil 840 843 841 844 case jmodels.CommitOperationDelete: ··· 855 858 856 859 tx, err := i.Db.Begin() 857 860 if err != nil { 858 858 - return err 861 861 + return fmt.Errorf("failed to start txn: %w", err) 859 862 } 860 863 defer func() { 861 864 tx.Rollback() ··· 869 872 orm.FilterEq("instance", instance), 870 873 ) 871 874 if err != nil { 872 872 - return err 875 875 + return fmt.Errorf("failed to remove spindle members: %w", err) 873 876 } 874 877 875 878 err = db.DeleteSpindle( ··· 878 881 orm.FilterEq("instance", instance), 879 882 ) 880 883 if err != nil { 881 881 - return err 884 884 + return fmt.Errorf("failed to delete spindle: %w", err) 882 885 } 883 886 884 887 if spindle.Verified != nil { 885 888 err = i.Enforcer.RemoveSpindle(instance) 886 889 if err != nil { 887 887 - return err 890 890 + return fmt.Errorf("failed to remove spindle from enforcer: %w", err) 888 891 } 889 892 } 890 893 891 894 err = tx.Commit() 892 895 if err != nil { 893 893 - return err 896 896 + return fmt.Errorf("failed to commit txn: %w", err) 894 897 } 895 898 896 899 err = i.Enforcer.E.SavePolicy() 897 900 if err != nil { 898 898 - return err 901 901 + return fmt.Errorf("failed to save ACLs: %w", err) 899 902 } 903 903 + 904 904 + l.Info("ingested record", "instance", instance) 900 905 } 901 906 902 907 return nil 903 908 } 904 909 905 905 - func (i *Ingester) ingestString(e *jmodels.Event) error { 910 910 + func (i *Ingester) ingestString(e *jmodels.Event, l *slog.Logger) error { 906 911 did := e.Did 907 912 rkey := e.Commit.RKey 908 913 909 914 var err error 910 915 911 911 - l := i.Logger.With("handler", "ingestString", "nsid", e.Commit.Collection, "did", did, "rkey", rkey) 912 912 - l.Info("ingesting record") 916 916 + l = l.With("handler", "ingestString") 913 917 914 918 switch e.Commit.Operation { 915 919 case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: ··· 933 937 return err 934 938 } 935 939 940 940 + l.Info("ingested record") 936 941 return nil 937 942 938 943 case jmodels.CommitOperationDelete: ··· 945 950 return fmt.Errorf("failed to delete string record: %w", err) 946 951 } 947 952 953 953 + l.Info("ingested record") 948 954 return nil 949 955 } 950 956 951 957 return nil 952 958 } 953 959 954 954 - func (i *Ingester) ingestKnotMember(ctx context.Context, e *jmodels.Event) error { 960 960 + func (i *Ingester) ingestKnotMember(ctx context.Context, e *jmodels.Event, l *slog.Logger) error { 955 961 did := e.Did 956 962 var err error 957 963 958 958 - l := i.Logger.With("handler", "ingestKnotMember") 959 959 - l = l.With("nsid", e.Commit.Collection) 964 964 + l = l.With("handler", "ingestKnotMember") 960 965 961 966 switch e.Commit.Operation { 962 967 case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: ··· 1117 1122 return nil 1118 1123 } 1119 1124 1120 1120 - func (i *Ingester) ingestKnot(ctx context.Context, e *jmodels.Event) error { 1125 1125 + func (i *Ingester) ingestKnot(ctx context.Context, e *jmodels.Event, l *slog.Logger) error { 1121 1126 did := e.Did 1122 1127 var err error 1123 1128 1124 1124 - l := i.Logger.With("handler", "ingestKnot") 1125 1125 - l = l.With("nsid", e.Commit.Collection) 1129 1129 + l = l.With("handler", "ingestKnot") 1126 1130 1127 1131 switch e.Commit.Operation { 1128 1132 case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: ··· 1146 1150 l.Warn("failed to verify knot", "domain", domain, "did", did, "err", err) 1147 1151 } 1148 1152 1153 1153 + l.Info("ingested record", "domain", domain) 1149 1154 return nil 1150 1155 1151 1156 case jmodels.CommitOperationDelete: ··· 1167 1172 1168 1173 tx, err := i.Db.Begin() 1169 1174 if err != nil { 1170 1170 - return err 1175 1175 + return fmt.Errorf("failed to start txn: %w", err) 1171 1176 } 1172 1177 defer func() { 1173 1178 tx.Rollback() ··· 1180 1185 orm.FilterEq("domain", domain), 1181 1186 ) 1182 1187 if err != nil { 1183 1183 - return err 1188 1188 + return fmt.Errorf("failed to remove knot members: %w", err) 1184 1189 } 1185 1190 1186 1191 err = db.DeleteKnot( ··· 1189 1194 orm.FilterEq("domain", domain), 1190 1195 ) 1191 1196 if err != nil { 1192 1192 - return err 1197 1197 + return fmt.Errorf("failed to delete knot: %w", err) 1193 1198 } 1194 1199 1195 1200 err = db.RemoveReposByKnot(tx, domain) 1196 1201 if err != nil { 1197 1197 - return err 1202 1202 + return fmt.Errorf("failed to remove repos by knot: %w", err) 1198 1203 } 1199 1204 1200 1205 if registration.Registered != nil { 1201 1206 err = i.Enforcer.RemoveKnot(domain) 1202 1207 if err != nil { 1203 1203 - return err 1208 1208 + return fmt.Errorf("failed to remove knot from enforcer: %w", err) 1204 1209 } 1205 1210 } 1206 1211 1207 1212 err = tx.Commit() 1208 1213 if err != nil { 1209 1209 - return err 1214 1214 + return fmt.Errorf("failed to commit txn: %w", err) 1210 1215 } 1211 1216 1212 1217 err = i.Enforcer.E.SavePolicy() 1213 1218 if err != nil { 1214 1214 - return err 1219 1219 + return fmt.Errorf("failed to save ACLs: %w", err) 1215 1220 } 1221 1221 + 1222 1222 + l.Info("ingested record", "domain", domain) 1216 1223 } 1217 1224 1218 1225 return nil ··· 1324 1331 g.Wait() 1325 1332 } 1326 1333 1327 1327 - func (i *Ingester) ingestIssue(ctx context.Context, e *jmodels.Event) error { 1334 1334 + func (i *Ingester) ingestIssue(ctx context.Context, e *jmodels.Event, l *slog.Logger) error { 1328 1335 did := e.Did 1329 1336 rkey := e.Commit.RKey 1330 1337 1331 1338 var err error 1332 1339 1333 1333 - l := i.Logger.With("handler", "ingestIssue", "nsid", e.Commit.Collection, "did", did, "rkey", rkey) 1334 1334 - l.Info("ingesting record") 1340 1340 + l = l.With("handler", "ingestIssue") 1335 1341 1336 1342 switch e.Commit.Operation { 1337 1343 case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: ··· 1384 1390 return err 1385 1391 } 1386 1392 1393 1393 + l.Info("ingested record") 1387 1394 return nil 1388 1395 1389 1396 case jmodels.CommitOperationDelete: ··· 1407 1414 return err 1408 1415 } 1409 1416 1417 1417 + l.Info("ingested record") 1410 1418 return nil 1411 1419 } 1412 1420 1413 1421 return nil 1414 1422 } 1415 1423 1416 1416 - func (i *Ingester) ingestPull(ctx context.Context, e *jmodels.Event) error { 1424 1424 + func (i *Ingester) ingestPull(ctx context.Context, e *jmodels.Event, l *slog.Logger) error { 1417 1425 did := e.Did 1418 1426 rkey := e.Commit.RKey 1419 1427 1420 1428 var err error 1421 1429 1422 1422 - l := i.Logger.With("handler", "ingestPull", "nsid", e.Commit.Collection, "did", did, "rkey", rkey) 1423 1423 - l.Info("ingesting record") 1430 1430 + l = l.With("handler", "ingestPull") 1424 1431 1425 1432 switch e.Commit.Operation { 1426 1433 case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: ··· 1434 1441 1435 1442 ownerId, err := i.IdResolver.ResolveIdent(ctx, did) 1436 1443 if err != nil { 1437 1437 - l.Error("failed to resolve did") 1444 1444 + l.Error("failed to resolve did", "err", err) 1438 1445 return err 1439 1446 } 1440 1447 ··· 1523 1530 return err 1524 1531 } 1525 1532 1533 1533 + l.Info("ingested record") 1526 1534 return nil 1527 1535 1528 1536 case jmodels.CommitOperationDelete: ··· 1546 1554 return err 1547 1555 } 1548 1556 1557 1557 + l.Info("ingested record") 1549 1558 return nil 1550 1559 } 1551 1560 ··· 1553 1562 } 1554 1563 1555 1564 // ingestIssueComment ingests legacy sh.tangled.repo.issue.comment deletions 1556 1556 - func (i *Ingester) ingestIssueComment(e *jmodels.Event) error { 1557 1557 - l := i.Logger.With("handler", "ingestIssueComment", "nsid", e.Commit.Collection, "did", e.Did, "rkey", e.Commit.RKey) 1558 1558 - l.Info("ingesting record") 1565 1565 + func (i *Ingester) ingestIssueComment(e *jmodels.Event, l *slog.Logger) error { 1566 1566 + l = l.With("handler", "ingestIssueComment") 1559 1567 1560 1568 switch e.Commit.Operation { 1561 1569 case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: ··· 1572 1580 } 1573 1581 } 1574 1582 1583 1583 + l.Info("ingested record") 1575 1584 return nil 1576 1585 } 1577 1586 1578 1587 // ingestPullComment ingests legacy sh.tangled.repo.pull.comment deletions 1579 1579 - func (i *Ingester) ingestPullComment(e *jmodels.Event) error { 1580 1580 - l := i.Logger.With("handler", "ingestPullComment", "nsid", e.Commit.Collection, "did", e.Did, "rkey", e.Commit.RKey) 1581 1581 - l.Info("ingesting record") 1588 1588 + func (i *Ingester) ingestPullComment(e *jmodels.Event, l *slog.Logger) error { 1589 1589 + l = l.With("handler", "ingestPullComment") 1582 1590 1583 1591 switch e.Commit.Operation { 1584 1592 case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: ··· 1595 1603 } 1596 1604 } 1597 1605 1606 1606 + l.Info("ingested record") 1598 1607 return nil 1599 1608 } 1600 1609 1601 1601 - func (i *Ingester) ingestComment(e *jmodels.Event) error { 1610 1610 + func (i *Ingester) ingestComment(e *jmodels.Event, l *slog.Logger) error { 1602 1611 did := e.Did 1603 1612 rkey := e.Commit.RKey 1604 1613 cid := e.Commit.CID 1605 1614 1606 1615 var err error 1607 1616 1608 1608 - l := i.Logger.With("handler", "ingestComment", "nsid", e.Commit.Collection, "did", did, "rkey", rkey) 1609 1609 - l.Info("ingesting record") 1617 1617 + l = l.With("handler", "ingestComment") 1610 1618 1611 1619 ctx := context.Background() 1612 1620 ··· 1657 1665 ); err != nil { 1658 1666 return fmt.Errorf("failed to delete comment record: %w", err) 1659 1667 } 1660 1660 - 1661 1661 - return nil 1662 1668 } 1663 1669 1670 1670 + l.Info("ingested record") 1664 1671 return nil 1665 1672 } 1666 1673 1667 1667 - func (i *Ingester) ingestReaction(e *jmodels.Event) error { 1674 1674 + func (i *Ingester) ingestReaction(e *jmodels.Event, l *slog.Logger) error { 1668 1675 did := e.Did 1669 1676 rkey := e.Commit.RKey 1670 1677 1671 1671 - l := i.Logger.With("handler", "ingestReaction", "nsid", e.Commit.Collection, "did", did, "rkey", rkey) 1672 1672 - l.Info("ingesting record") 1678 1678 + l = l.With("handler", "ingestReaction") 1673 1679 1674 1680 switch e.Commit.Operation { 1675 1681 case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: ··· 1708 1714 return fmt.Errorf("failed to add reaction: %w", err) 1709 1715 } 1710 1716 1711 1711 - return tx.Commit() 1717 1717 + if err := tx.Commit(); err != nil { 1718 1718 + return err 1719 1719 + } 1712 1720 1713 1721 case jmodels.CommitOperationDelete: 1714 1722 if err := db.DeleteReactionByRkey(i.Db, did, rkey); err != nil { ··· 1716 1724 } 1717 1725 } 1718 1726 1727 1727 + l.Info("ingested record") 1719 1728 return nil 1720 1729 } 1721 1730 1722 1722 - func (i *Ingester) ingestLabelDefinition(e *jmodels.Event) error { 1731 1731 + func (i *Ingester) ingestLabelDefinition(e *jmodels.Event, l *slog.Logger) error { 1723 1732 did := e.Did 1724 1733 rkey := e.Commit.RKey 1725 1734 1726 1735 var err error 1727 1736 1728 1728 - l := i.Logger.With("handler", "ingestLabelDefinition", "nsid", e.Commit.Collection, "did", did, "rkey", rkey) 1729 1729 - l.Info("ingesting record") 1737 1737 + l = l.With("handler", "ingestLabelDefinition") 1730 1738 1731 1739 switch e.Commit.Operation { 1732 1740 case jmodels.CommitOperationCreate, jmodels.CommitOperationUpdate: ··· 1751 1759 return fmt.Errorf("failed to create labeldef: %w", err) 1752 1760 } 1753 1761 1762 1762 + l.Info("ingested record") 1754 1763 return nil 1755 1764 1756 1765 case jmodels.CommitOperationDelete: ··· 1762 1771 return fmt.Errorf("failed to delete labeldef record: %w", err) 1763 1772 } 1764 1773 1774 1774 + l.Info("ingested record") 1765 1775 return nil 1766 1776 } 1767 1777 1768 1778 return nil 1769 1779 } 1770 1780 1771 1771 - func (i *Ingester) ingestLabelOp(e *jmodels.Event) error { 1781 1781 + func (i *Ingester) ingestLabelOp(e *jmodels.Event, l *slog.Logger) error { 1772 1782 did := e.Did 1773 1783 rkey := e.Commit.RKey 1774 1784 1775 1785 var err error 1776 1786 1777 1777 - l := i.Logger.With("handler", "ingestLabelOp", "nsid", e.Commit.Collection, "did", did, "rkey", rkey) 1778 1778 - l.Info("ingesting record") 1787 1787 + l = l.With("handler", "ingestLabelOp") 1779 1788 1780 1789 switch e.Commit.Operation { 1781 1790 case jmodels.CommitOperationCreate: ··· 1840 1849 if err = tx.Commit(); err != nil { 1841 1850 return err 1842 1851 } 1852 1852 + 1853 1853 + l.Info("ingested record") 1843 1854 } 1844 1855 1845 1856 return nil

+12 -12

appview/ingester_repo.go

Reviewed

··· 19 19 "tangled.org/core/orm" 20 20 ) 21 21 22 22 - func (i *Ingester) ingestRepo(ctx context.Context, e *jmodels.Event) error { 23 23 - l := i.Logger.With("handler", "ingestRepo", "did", e.Did, "rkey", e.Commit.RKey) 22 22 + func (i *Ingester) ingestRepo(ctx context.Context, e *jmodels.Event, l *slog.Logger) error { 23 23 + l = l.With("handler", "ingestRepo") 24 24 25 25 switch e.Commit.Operation { 26 26 case jmodels.CommitOperationCreate: 27 27 - return i.ingestRepoCreate(ctx, e) 27 27 + return i.ingestRepoCreate(ctx, e, l) 28 28 case jmodels.CommitOperationUpdate: 29 29 - return i.ingestRepoUpdate(ctx, e) 29 29 + return i.ingestRepoUpdate(ctx, e, l) 30 30 case jmodels.CommitOperationDelete: 31 31 - return i.ingestRepoDelete(ctx, e) 31 31 + return i.ingestRepoDelete(ctx, e, l) 32 32 default: 33 33 - l.Info("unknown repo operation", "op", e.Commit.Operation) 33 33 + l.Info("unknown repo operation") 34 34 return nil 35 35 } 36 36 } 37 37 38 38 - func (i *Ingester) ingestRepoCreate(ctx context.Context, e *jmodels.Event) error { 39 39 - l := i.Logger.With("handler", "ingestRepoCreate", "did", e.Did, "rkey", e.Commit.RKey) 38 38 + func (i *Ingester) ingestRepoCreate(ctx context.Context, e *jmodels.Event, l *slog.Logger) error { 39 39 + l = l.With("handler", "ingestRepoCreate") 40 40 41 41 record := tangled.Repo{} 42 42 if err := json.Unmarshal(json.RawMessage(e.Commit.Record), &record); err != nil { ··· 182 182 return i.Enforcer.E.SavePolicy() 183 183 } 184 184 185 185 - func (i *Ingester) ingestRepoUpdate(ctx context.Context, e *jmodels.Event) error { 186 186 - l := i.Logger.With("handler", "ingestRepoUpdate", "did", e.Did, "rkey", e.Commit.RKey) 185 185 + func (i *Ingester) ingestRepoUpdate(ctx context.Context, e *jmodels.Event, l *slog.Logger) error { 186 186 + l = l.With("handler", "ingestRepoUpdate") 187 187 188 188 record := tangled.Repo{} 189 189 if err := json.Unmarshal(json.RawMessage(e.Commit.Record), &record); err != nil { ··· 247 247 return tx.Commit() 248 248 } 249 249 250 250 - func (i *Ingester) ingestRepoDelete(ctx context.Context, e *jmodels.Event) error { 251 251 - l := i.Logger.With("handler", "ingestRepoDelete", "did", e.Did, "rkey", e.Commit.RKey) 250 250 + func (i *Ingester) ingestRepoDelete(ctx context.Context, e *jmodels.Event, l *slog.Logger) error { 251 251 + l = l.With("handler", "ingestRepoDelete") 252 252 253 253 repo, err := db.GetRepo(i.Db, 254 254 orm.FilterEq("did", e.Did),

+12 -13

appview/ingester_repo_test.go

Reviewed

··· 5 5 "database/sql" 6 6 "encoding/json" 7 7 "errors" 8 8 - "io" 9 8 "log/slog" 10 9 "net/url" 11 10 "path/filepath" ··· 79 78 ing := &Ingester{ 80 79 Db: d, 81 80 Enforcer: enforcer, 82 82 - Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), 81 81 + Logger: slog.New(slog.DiscardHandler), 83 82 Notifier: spy, 84 83 } 85 84 return ing, spy ··· 93 92 func ingestAcceptingOwner(t *testing.T, ing *Ingester, e *jmodels.Event) error { 94 93 t.Helper() 95 94 ing.Verifier = acceptOwner(t, e) 96 96 - return ing.ingestRepo(context.Background(), e) 95 95 + return ing.ingestRepo(context.Background(), e, ing.Logger) 97 96 } 98 97 99 98 func seedRepoRow(t *testing.T, ing *Ingester, did, knot, name, rkey, repoDid string) *models.Repo { ··· 566 565 KnotURL: mustKnotURL(t, "https://knot.example"), 567 566 }, nil)) 568 567 569 569 - if err := ing.ingestRepo(context.Background(), e); err != nil { 568 568 + if err := ing.ingestRepo(context.Background(), e, ing.Logger); err != nil { 570 569 t.Fatalf("ingestRepo: %v", err) 571 570 } 572 571 ··· 596 595 KnotURL: mustKnotURL(t, "https://knot.example"), 597 596 }, nil)) 598 597 599 599 - if err := ing.ingestRepo(context.Background(), e); err != nil { 598 598 + if err := ing.ingestRepo(context.Background(), e, ing.Logger); err != nil { 600 599 t.Fatalf("ingestRepo: %v", err) 601 600 } 602 601 ··· 625 624 KnotURL: mustKnotURL(t, "https://knot.example"), 626 625 }, nil)) 627 626 628 628 - if err := ing.ingestRepo(context.Background(), e); err != nil { 627 627 + if err := ing.ingestRepo(context.Background(), e, ing.Logger); err != nil { 629 628 t.Fatalf("ingestRepo: %v", err) 630 629 } 631 630 ··· 648 647 649 648 withVerifier(ing, stubVerifier(repoverify.Result{}, errors.New("knot unreachable"))) 650 649 651 651 - err := ing.ingestRepo(context.Background(), e) 650 650 + err := ing.ingestRepo(context.Background(), e, ing.Logger) 652 651 if err == nil { 653 652 t.Fatalf("expected error on transient verifier failure, got nil") 654 653 } ··· 673 672 KnotURL: mustKnotURL(t, "https://knot.example"), 674 673 }, nil)) 675 674 676 676 - if err := ing.ingestRepo(context.Background(), e); err != nil { 675 675 + if err := ing.ingestRepo(context.Background(), e, ing.Logger); err != nil { 677 676 t.Fatalf("ingestRepo: %v", err) 678 677 } 679 678 ··· 697 696 return repoverify.Result{}, nil 698 697 }) 699 698 700 700 - if err := ing.ingestRepo(context.Background(), e); err != nil { 699 699 + if err := ing.ingestRepo(context.Background(), e, ing.Logger); err != nil { 701 700 t.Fatalf("ingestRepo: %v", err) 702 701 } 703 702 if verifierCalled { ··· 716 715 RepoDid: ptr("did:plc:akshays-repo"), 717 716 }) 718 717 719 719 - err := ing.ingestRepo(context.Background(), e) 718 718 + err := ing.ingestRepo(context.Background(), e, ing.Logger) 720 719 if err == nil { 721 720 t.Fatalf("expected error when Verifier is nil, got nil") 722 721 } ··· 739 738 KnotURL: mustKnotURL(t, "https://knot.example"), 740 739 }, nil)) 741 740 742 742 - if err := ing.ingestRepo(context.Background(), e); err != nil { 741 741 + if err := ing.ingestRepo(context.Background(), e, ing.Logger); err != nil { 743 742 t.Fatalf("ingestRepo: %v", err) 744 743 } 745 744 if _, err := db.GetRepo(ing.Db, ··· 769 768 KnotURL: mustKnotURL(t, "https://knot.example"), 770 769 }, nil)) 771 770 772 772 - if err := ing.ingestRepo(context.Background(), e); err != nil { 771 771 + if err := ing.ingestRepo(context.Background(), e, ing.Logger); err != nil { 773 772 t.Fatalf("ingestRepo: %v", err) 774 773 } 775 774 akshay := loadRepo(t, ing, "did:plc:akshay", "akshayskey") ··· 797 796 KnotURL: mustKnotURL(t, "https://knot.example"), 798 797 }, nil)) 799 798 800 800 - if err := ing.ingestRepo(context.Background(), e); err != nil { 799 799 + if err := ing.ingestRepo(context.Background(), e, ing.Logger); err != nil { 801 800 t.Fatalf("ingestRepo: %v", err) 802 801 } 803 802 akshay := loadRepo(t, ing, "did:plc:akshay", "akshayskey")

+11 -12

appview/ingester_string_test.go

Reviewed

··· 2 2 3 3 import ( 4 4 "encoding/json" 5 5 - "io" 6 5 "log/slog" 7 6 "path/filepath" 8 7 "strings" ··· 27 26 t.Cleanup(func() { d.Close() }) 28 27 return &Ingester{ 29 28 Db: d, 30 30 - Logger: slog.New(slog.NewTextHandler(io.Discard, nil)), 29 29 + Logger: slog.New(slog.DiscardHandler), 31 30 Validator: &validator.Validator{}, 32 31 } 33 32 } ··· 79 78 CreatedAt: created.Format(time.RFC3339), 80 79 }) 81 80 82 82 - if err := ing.ingestString(e); err != nil { 81 81 + if err := ing.ingestString(e, ing.Logger); err != nil { 83 82 t.Fatalf("ingestString: %v", err) 84 83 } 85 84 ··· 114 113 CreatedAt: created.Format(time.RFC3339), 115 114 } 116 115 117 117 - if err := ing.ingestString(makeStringEvent(t, jmodels.CommitOperationCreate, "did:plc:boltless", "rk1", base)); err != nil { 116 116 + if err := ing.ingestString(makeStringEvent(t, jmodels.CommitOperationCreate, "did:plc:boltless", "rk1", base), ing.Logger); err != nil { 118 117 t.Fatalf("ingestString create: %v", err) 119 118 } 120 119 121 120 updated := base 122 121 updated.Contents = "hello, world!\n" 123 123 - if err := ing.ingestString(makeStringEvent(t, jmodels.CommitOperationUpdate, "did:plc:boltless", "rk1", updated)); err != nil { 122 122 + if err := ing.ingestString(makeStringEvent(t, jmodels.CommitOperationUpdate, "did:plc:boltless", "rk1", updated), ing.Logger); err != nil { 124 123 t.Fatalf("ingestString update: %v", err) 125 124 } 126 125 ··· 148 147 CreatedAt: time.Date(2025, 9, 14, 10, 30, 0, 0, time.UTC).Format(time.RFC3339), 149 148 } 150 149 151 151 - if err := ing.ingestString(makeStringEvent(t, jmodels.CommitOperationCreate, "did:plc:akshay", "rk2", rec)); err != nil { 150 150 + if err := ing.ingestString(makeStringEvent(t, jmodels.CommitOperationCreate, "did:plc:akshay", "rk2", rec), ing.Logger); err != nil { 152 151 t.Fatalf("create: %v", err) 153 152 } 154 154 - if err := ing.ingestString(makeStringEvent(t, jmodels.CommitOperationUpdate, "did:plc:akshay", "rk2", rec)); err != nil { 153 153 + if err := ing.ingestString(makeStringEvent(t, jmodels.CommitOperationUpdate, "did:plc:akshay", "rk2", rec), ing.Logger); err != nil { 155 154 t.Fatalf("update: %v", err) 156 155 } 157 156 ··· 168 167 Contents: "x", 169 168 CreatedAt: time.Now().UTC().Format(time.RFC3339), 170 169 } 171 171 - if err := ing.ingestString(makeStringEvent(t, jmodels.CommitOperationCreate, "did:plc:boltless", "rk1", rec)); err != nil { 170 170 + if err := ing.ingestString(makeStringEvent(t, jmodels.CommitOperationCreate, "did:plc:boltless", "rk1", rec), ing.Logger); err != nil { 172 171 t.Fatalf("create: %v", err) 173 172 } 174 173 ··· 181 180 RKey: "rk1", 182 181 }, 183 182 } 184 184 - if err := ing.ingestString(del); err != nil { 183 183 + if err := ing.ingestString(del, ing.Logger); err != nil { 185 184 t.Fatalf("delete: %v", err) 186 185 } 187 186 ··· 206 205 for _, tc := range cases { 207 206 t.Run(tc.name, func(t *testing.T) { 208 207 e := makeStringEvent(t, jmodels.CommitOperationCreate, "did:plc:akshay", "bad", tc.rec) 209 209 - if err := ing.ingestString(e); err == nil { 208 208 + if err := ing.ingestString(e, ing.Logger); err == nil { 210 209 t.Fatal("expected validator error, got nil") 211 210 } 212 211 if _, ok := loadString(t, ing, "did:plc:akshay", "bad"); ok { ··· 227 226 event := makeStringEvent(t, jmodels.CommitOperationCreate, "did:plc:boltless", "rkcold", rec) 228 227 229 228 first := newStringIngester(t) 230 230 - if err := first.ingestString(event); err != nil { 229 229 + if err := first.ingestString(event, first.Logger); err != nil { 231 230 t.Fatalf("first ingest: %v", err) 232 231 } 233 232 live, _ := loadString(t, first, "did:plc:boltless", "rkcold") 234 233 235 234 second := newStringIngester(t) 236 236 - if err := second.ingestString(event); err != nil { 235 235 + if err := second.ingestString(event, second.Logger); err != nil { 237 236 t.Fatalf("replay ingest: %v", err) 238 237 } 239 238 replayed, _ := loadString(t, second, "did:plc:boltless", "rkcold")