1package daemon23import (4 "context"5 "crypto/sha256"6 "database/sql"7 "encoding/hex"8 "encoding/json"9 "errors"10 "net/url"11 "sort"12 "strings"13 "time"1415 "github.com/FelineStateMachine/tinyrelay/internal/event"16 gitrelay "github.com/FelineStateMachine/tinyrelay/tinygit"17 "github.com/FelineStateMachine/tinyrelay/internal/replication"18 "github.com/FelineStateMachine/tinyrelay/internal/storage"19 "github.com/FelineStateMachine/tinyrelay/internal/syncprotocol"20)2122// gitSync repairs signed branch state. Conversation history is scheduled23// separately so a missing Git object cannot block issue or PR replication.24func (t *Tenant) gitSync(ctx context.Context, repo gitrelay.Repository) error {25 if len(repo.Refs) == 0 {26 return nil27 }28 repo.Private = repo.Private || t.PrivateServiceEnabled()29 t.git.ConfigurePrivatePeers(t.Policy().PrivatePeers, t.privateHTTPAuth)30 return t.git.FetchMissing(ctx, repo, gitrelay.RepairSources(repo), repo.Refs)31}3233func (t *Tenant) gitEventSync(ctx context.Context, repo gitrelay.Repository) error {34 var result error35 targets := boundedGitRelays(repo.Relays, repo.Private || t.PrivateServiceEnabled(), t.Policy().PrivatePeers)36 setting := "grasp02.peer-offset." + repo.Owner + ":" + repo.Identifier37 var offset int38 _ = t.store.GetSetting(ctx, setting, &offset)39 if offset < 0 {40 offset = 041 }42 targets = rotateGitRelays(targets, offset)43 historyCtx, historyCancel := context.WithTimeout(ctx, 20*time.Second)44 for index, target := range targets {45 if historyCtx.Err() != nil {46 break47 }48 if sameRelayGit(target, t.RelayURL(), t.publicURL) {49 continue50 }51 // Persist the next starting peer before a bounded network attempt.52 // An unavailable peer cannot monopolize every subsequent pass.53 result = errors.Join(result, t.store.PutSetting(ctx, setting, (offset+index+1)%len(targets)))54 peerCtx, cancel := context.WithTimeout(historyCtx, 8*time.Second)55 result = errors.Join(result, t.gitSyncPeer(peerCtx, repo, target))56 cancel()57 }58 if historyCtx.Err() != nil && ctx.Err() == nil {59 result = errors.Join(result, replication.ErrPullIncomplete)60 }61 historyCancel()62 repairCtx, repairCancel := context.WithTimeout(ctx, 5*time.Second)63 result = errors.Join(result, t.repairGitPullRequests(repairCtx, repo))64 repairCancel()65 if t.Policy().Features.Grasp03 {66 outboxCtx, outboxCancel := context.WithTimeout(ctx, 15*time.Second)67 result = errors.Join(result, t.gitOutboxSync(outboxCtx, repo))68 outboxCancel()69 }70 return gitSyncOutcome(ctx, result)71}7273// gitSyncOutcome classifies a pass for the scheduler. Deadlines from the74// pass's own time budgets are dropped because peer and outbox offsets persist75// between passes. A pass whose only remaining condition is more history to76// fetch reports gitrelay.ErrIncomplete so it continues soon without counting77// as a failure. Cancellation of the caller's context still propagates.78func gitSyncOutcome(parent context.Context, err error) error {79 err = withoutBudgetDeadline(parent, err)80 if err == nil {81 return nil82 }83 if onlyPullIncomplete(err) {84 return gitrelay.ErrIncomplete85 }86 return err87}8889func withoutBudgetDeadline(parent context.Context, err error) error {90 if err == nil || parent.Err() != nil {91 return err92 }93 if joined, ok := err.(interface{ Unwrap() []error }); ok {94 var kept []error95 for _, item := range joined.Unwrap() {96 if item = withoutBudgetDeadline(parent, item); item != nil {97 kept = append(kept, item)98 }99 }100 return errors.Join(kept...)101 }102 if errors.Is(err, context.DeadlineExceeded) {103 return nil104 }105 return err106}107108func onlyPullIncomplete(err error) bool {109 if joined, ok := err.(interface{ Unwrap() []error }); ok {110 for _, item := range joined.Unwrap() {111 if !onlyPullIncomplete(item) {112 return false113 }114 }115 return true116 }117 return errors.Is(err, replication.ErrPullIncomplete)118}119120// gitIngestRejected reports a deterministic admission decision. NIP-01121// prefixes mark policy rejections that repeat on every attempt; other errors122// are storage or lookup failures that may succeed later.123func gitIngestRejected(err error) bool {124 message := err.Error()125 for _, prefix := range []string{"blocked:", "invalid:", "restricted:", "auth-required:", "pow:"} {126 if strings.HasPrefix(message, prefix) {127 return true128 }129 }130 return false131}132133// gitTargetTransport checks the privacy boundary before sending any filters.134// An unverified private peer never falls back to an anonymous connection.135func (t *Tenant) gitTargetTransport(ctx context.Context, repo gitrelay.Repository, target string) (*replication.NostrTransport, error) {136 p := t.Policy()137 if !p.Features.Grasp || !p.Features.Grasp02 {138 return nil, errors.New("GRASP synchronization is disabled")139 }140 if !gitOutboxAllowedRelay(target, repo.Private || t.PrivateServiceEnabled(), p.PrivatePeers) {141 return nil, errors.New("GRASP synchronization target is not allowed")142 }143 if repo.Private || t.PrivateServiceEnabled() {144 if !t.PrivateServiceEnabled() || !privatePeerMatch(target, p.PrivatePeers) || !t.privatePeerReady(ctx, target) {145 return nil, errors.New("private peer is not ready")146 }147 }148 return &replication.NostrTransport{Dialer: t.gitEventDialer(repo.Private || t.PrivateServiceEnabled()), Timeout: 5 * time.Second, LegacyCache: t.gitLegacy}, nil149}150151func (t *Tenant) gitSyncPeer(ctx context.Context, repo gitrelay.Repository, target string) error {152 transport, err := t.gitTargetTransport(ctx, repo, target)153 if err != nil {154 return err155 }156 var result error157 cursors := make(map[string]struct{})158 for _, filter := range gitEventFilters(repo) {159 cursors[gitHistoryCursorKey(target, filter)] = struct{}{}160 result = errors.Join(result, t.gitPullFilter(ctx, transport, target, filter))161 }162 events, err := t.gitConversationEvents(ctx, repo)163 if err != nil {164 return errors.Join(result, err)165 }166 for _, filter := range gitReplyFilters(events) {167 cursors[gitHistoryCursorKey(target, filter)] = struct{}{}168 result = errors.Join(result, t.gitPullFilter(ctx, transport, target, filter))169 }170 // Reply filters are chunked by root ID, so their keys shift as roots171 // arrive. Drop cursors for filters this pass no longer uses.172 return errors.Join(result, t.store.PruneSettings(ctx, gitHistoryCursorPrefix(target), cursors))173}174175func (t *Tenant) gitPullFilter(ctx context.Context, transport *replication.NostrTransport, target string, filter event.Filter) error {176 local, err := t.localSyncItems(ctx, filter)177 if errors.Is(err, replication.ErrPullIncomplete) {178 return t.gitPullLargeFilter(ctx, transport, target, filter)179 }180 if err != nil {181 return err182 }183 items, pullErr := transport.QuerySynchronized(ctx, target, filter, local)184 return errors.Join(pullErr, t.ingestGitHistory(ctx, filter, items))185}186187// A partial inventory is not safe for negentropy: the resulting missing-ID set188// can overflow its fetch cap. Refresh a small head first, then resume an older189// window. A slow historical query cannot prevent new events from being stored.190func (t *Tenant) gitPullLargeFilter(ctx context.Context, transport *replication.NostrTransport, target string, filter event.Filter) error {191 olderFilter, hasCursor, err := t.gitHistoryWindow(ctx, target, filter)192 if err != nil {193 return err194 }195 headCtx, cancelHead := gitQueryContext(ctx, 2*time.Second)196 head, headErr := queryGitHead(headCtx, transport, target, filter)197 cancelHead()198 headIngestErr := t.ingestGitHistory(ctx, filter, head)199 result := errors.Join(replication.ErrPullIncomplete, headErr, headIngestErr)200 key := gitHistoryCursorKey(target, filter)201 if !hasCursor {202 // A completed bounded head query seeds history; it never proves that203 // the remote inventory is exhausted. Failed queries cannot move a204 // timestamp cursor because events may have arrived out of order.205 if headErr == nil && headIngestErr == nil && len(head) > 0 {206 result = errors.Join(result, t.store.PutSetting(ctx, key, oldestGitHistoryCursor(head)))207 }208 return result209 }210 if ctx.Err() != nil {211 return errors.Join(result, ctx.Err())212 }213 olderCtx, cancelOlder := gitQueryContext(ctx, 5*time.Second)214 older, olderErr := transport.QueryPaginated(olderCtx, target, olderFilter)215 cancelOlder()216 olderIngestErr := t.ingestGitHistory(ctx, filter, older)217 result = errors.Join(result, olderErr, olderIngestErr)218 if olderIngestErr != nil {219 return result220 }221 if olderErr == nil {222 // Finished this sweep. The next pass starts another from the head,223 // allowing older events newly acquired by the peer to be discovered.224 result = errors.Join(result, t.store.DeleteSetting(ctx, key))225 } else if errors.Is(olderErr, replication.ErrPullIncomplete) && len(older) > 0 {226 result = errors.Join(result, t.store.PutSetting(ctx, key, oldestGitHistoryCursor(older)))227 }228 return result229}230231func (t *Tenant) ingestGitHistory(ctx context.Context, filter event.Filter, items []event.Event) error {232 sort.SliceStable(items, func(i, j int) bool { return metadataEventOrder(items[i].Kind) < metadataEventOrder(items[j].Kind) })233 var result error234 for _, item := range items {235 if ctx.Err() != nil {236 return errors.Join(result, ctx.Err())237 }238 if !event.Matches(filter, item) || event.Validate(item) != nil {239 continue240 }241 // A rejected event cannot prevent later valid events from arriving,242 // and a policy rejection repeats on every attempt, so it counts as243 // processed. A storage failure keeps the checkpoint so the window is244 // retried without skipping an event that was not persisted.245 if err := t.ingest(ctx, item, replication.OriginImport); err != nil && !errors.Is(err, storage.ErrDuplicate) && !errors.Is(err, storage.ErrReplaced) {246 t.app.telemetry.Logger().Debug("Git sync event rejected", "kind", item.Kind, "error", err)247 if !gitIngestRejected(err) {248 result = errors.Join(result, err)249 }250 }251 }252 return result253}254255const gitHeadPageSize = 256256257// Reserve time to ingest the response using the parent context. A query that258// expires may still have returned valid events worth retaining for the retry.259func gitQueryContext(ctx context.Context, maximum time.Duration) (context.Context, context.CancelFunc) {260 if deadline, ok := ctx.Deadline(); ok {261 if remaining := time.Until(deadline) / 2; remaining < maximum {262 maximum = remaining263 }264 }265 return context.WithTimeout(ctx, maximum)266}267268func queryGitHead(ctx context.Context, transport replication.PullTransport, target string, filter event.Filter) ([]event.Event, error) {269 limit := gitHeadPageSize270 filter.Limit = &limit271 return transport.Query(ctx, target, filter)272}273274func oldestGitHistoryCursor(items []event.Event) storage.EventCursor {275 if len(items) == 0 {276 return storage.EventCursor{}277 }278 oldest := items[0]279 for _, item := range items[1:] {280 if item.CreatedAt < oldest.CreatedAt || (item.CreatedAt == oldest.CreatedAt && item.ID > oldest.ID) {281 oldest = item282 }283 }284 return storage.EventCursor{CreatedAt: oldest.CreatedAt, ID: oldest.ID}285}286287func gitHistoryCursorPrefix(target string) string { return "grasp02.history." + target + "." }288289func gitHistoryCursorKey(target string, filter event.Filter) string {290 raw, _ := json.Marshal(filter)291 sum := sha256.Sum256(raw)292 return gitHistoryCursorPrefix(target) + hex.EncodeToString(sum[:])293}294295func (t *Tenant) gitHistoryWindow(ctx context.Context, target string, filter event.Filter) (event.Filter, bool, error) {296 var cursor storage.EventCursor297 err := t.store.GetSetting(ctx, gitHistoryCursorKey(target, filter), &cursor)298 if err != nil && !errors.Is(err, sql.ErrNoRows) {299 return filter, false, err300 }301 if cursor.CreatedAt > 0 {302 until := cursor.CreatedAt303 filter.Until = &until304 return filter, true, nil305 }306 return filter, false, nil307}308309func (t *Tenant) localSyncItems(ctx context.Context, filter event.Filter) ([]syncprotocol.Item, error) {310 rows, err := t.store.Query(ctx, filter, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}, Limit: 10000})311 if err != nil {312 return nil, err313 }314 // A bounded inventory is intentionally reported as incomplete. The caller315 // switches to a bounded newest-first pull so a large local history cannot316 // overflow negentropy's missing-ID fetch cap.317 items := make([]syncprotocol.Item, 0, len(rows.Events))318 for _, item := range rows.Events {319 items = append(items, syncprotocol.Item{ID: item.ID, Timestamp: item.CreatedAt})320 }321 if rows.More {322 return items, replication.ErrPullIncomplete323 }324 return items, nil325}326327func (t *Tenant) gitConversationEvents(ctx context.Context, repo gitrelay.Repository) ([]event.Event, error) {328 var items []event.Event329 for _, filter := range gitEventFilters(repo)[1:] {330 rows, err := t.store.Query(ctx, filter, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}, Limit: 10000})331 if err != nil {332 return nil, err333 }334 items = append(items, rows.Events...)335 }336 roots := uniqueEvents(items)337 for _, filter := range gitReplyFilters(roots) {338 rows, err := t.store.Query(ctx, filter, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}, Limit: 10000})339 if err != nil {340 return nil, err341 }342 items = append(items, rows.Events...)343 }344 return uniqueEvents(items), nil345}346347func gitReplyFilters(items []event.Event) []event.Filter {348 var ids []string349 for _, item := range items {350 if item.Kind == 1617 || item.Kind == 1618 || item.Kind == 1621 {351 ids = append(ids, item.ID)352 }353 }354 var filters []event.Filter355 for start := 0; start < len(ids); start += 100 {356 end := start + 100357 if end > len(ids) {358 end = len(ids)359 }360 for _, key := range []string{"e", "E"} {361 filters = append(filters, event.Filter{Kinds: []int{1619, 1111, 1630, 1631, 1632, 1633}, Tags: map[string][]string{key: ids[start:end]}})362 }363 }364 return filters365}366367func (t *Tenant) repairGitPullRequests(ctx context.Context, repo gitrelay.Repository) error {368 repo.Private = repo.Private || t.PrivateServiceEnabled()369 items, err := t.gitConversationEvents(ctx, repo)370 if err != nil {371 return err372 }373 t.git.ConfigurePrivatePeers(t.Policy().PrivatePeers, t.privateHTTPAuth)374 var result error375 for _, root := range items {376 if root.Kind != 1618 {377 continue378 }379 set := gitrelay.PullRequestRepair{Root: root}380 for _, update := range items {381 if update.Kind == 1619 && update.PubKey == root.PubKey && event.Tag(update, "E") == root.ID && event.Tag(update, "P") == root.PubKey {382 set.Updates = append(set.Updates, update)383 }384 }385 err := t.git.RepairPullRequestObjects(ctx, repo, set)386 result = errors.Join(result, err)387 if ctx.Err() != nil {388 return errors.Join(result, ctx.Err())389 }390 }391 return result392}393394func metadataEventOrder(kind int) int {395 if kind == event.KIND_REPO {396 return 0397 }398 return 1399}400401func (t *Tenant) gitEventDialer(private bool) replication.WebsocketDialer {402 dialer := replication.WebsocketDialer{AllowPrivate: t.app.cfg.AllowPrivateRelays, MaxMessageBytes: t.app.cfg.MaxMessageBytes}403 if !private {404 return dialer405 }406 dialer.Authenticate = func(authCtx context.Context, relayURL, challenge string) (event.Event, error) {407 if !t.PrivateServiceEnabled() || !privatePeerMatch(relayURL, t.Policy().PrivatePeers) {408 return event.Event{}, errors.New("private peer: authentication is no longer authorized")409 }410 return t.records.SignNIP42(authCtx, relayURL, challenge, 0)411 }412 return dialer413}414415func gitEventFilters(repo gitrelay.Repository) []event.Filter {416 authors := append([]string{repo.Owner}, repo.Maintainers...)417 metadataKinds := []int{event.KIND_REPO, event.KIND_REPO_STATE}418 collaborationKinds := []int{event.KIND_GIT_PATCH, event.KIND_GIT_PR, event.KIND_GIT_PR_UPDATE, event.KIND_GIT_ISSUE, 1111, 1630, 1631, 1632, 1633}419 coordinate := "30617:" + repo.Owner + ":" + repo.Identifier420 return []event.Filter{421 {Kinds: metadataKinds, Authors: authors, Tags: map[string][]string{"d": {repo.Identifier}}},422 {Kinds: collaborationKinds, Tags: map[string][]string{"a": {coordinate}}},423 {Kinds: collaborationKinds, Tags: map[string][]string{"A": {coordinate}}},424 }425}426427func sameRelayGit(a, b, c string) bool {428 a, b, c = normalizeRelayGit(a), normalizeRelayGit(b), normalizeRelayGit(c)429 return a != "" && (a == b || a == c)430}431432func normalizeRelayGit(raw string) string {433 raw = strings.TrimRight(strings.TrimSpace(raw), "/")434 u, err := url.Parse(raw)435 if err != nil {436 return raw437 }438 switch u.Scheme {439 case "http":440 u.Scheme = "ws"441 case "https":442 u.Scheme = "wss"443 }444 u.Host = strings.ToLower(u.Host)445 return strings.TrimRight(u.String(), "/")446}447448func uniqueEvents(rows []event.Event) []event.Event {449 seen := make(map[string]struct{}, len(rows))450 result := make([]event.Event, 0, len(rows))451 for _, row := range rows {452 if _, ok := seen[row.ID]; ok {453 continue454 }455 seen[row.ID] = struct{}{}456 result = append(result, row)457 }458 return result459}460461// sortEventsNewestFirst matches the stable storage order used by cursors.462func sortEventsNewestFirst(rows []event.Event) {463 sort.Slice(rows, func(i, j int) bool {464 if rows[i].CreatedAt == rows[j].CreatedAt {465 return rows[i].ID < rows[j].ID466 }467 return rows[i].CreatedAt > rows[j].CreatedAt468 })469}