| 1 | package tinygit |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/sha256" |
| 6 | "encoding/hex" |
| 7 | "encoding/json" |
| 8 | "errors" |
| 9 | "fmt" |
| 10 | "os" |
| 11 | "path/filepath" |
| 12 | "sort" |
| 13 | "strconv" |
| 14 | "strings" |
| 15 | "time" |
| 16 | |
| 17 | event "github.com/FelineStateMachine/tinyrelay/protocol/nostr" |
| 18 | ) |
| 19 | |
| 20 | // Service is the durable GRASP scheduler boundary. Tick is safe to call from |
| 21 | // a daemon ticker and runs one pass over the currently accepted repositories. |
| 22 | // Callback progress is persisted so a restart does not lose the last pass. |
| 23 | type Service struct { |
| 24 | relay *GitRelay |
| 25 | } |
| 26 | |
| 27 | // ValidateImported admits metadata received through an explicitly |
| 28 | // synchronized relay. Local announcements continue through Validate. |
| 29 | func (g *GitRelay) ValidateImported(ctx context.Context, e event.Event) (Repository, error) { |
| 30 | if err := event.Validate(e); err != nil { |
| 31 | return Repository{}, err |
| 32 | } |
| 33 | if e.Kind != 30617 && e.Kind != 30618 { |
| 34 | return Repository{}, fmt.Errorf("unsupported: event kind %d" , e.Kind) |
| 35 | } |
| 36 | r, err := g.parseRepository(e) |
| 37 | if err != nil { |
| 38 | return Repository{}, err |
| 39 | } |
| 40 | if g.authorize != nil { |
| 41 | if err := g.authorize(ctx, e, r); err != nil { |
| 42 | return Repository{}, err |
| 43 | } |
| 44 | } |
| 45 | if e.Kind == 30618 { |
| 46 | if r.EventID == "" { |
| 47 | return Repository{}, errors.New("blocked: repository announcement is missing" ) |
| 48 | } |
| 49 | if err := g.validateRefs(r.Refs); err != nil { |
| 50 | return Repository{}, err |
| 51 | } |
| 52 | } |
| 53 | return r, nil |
| 54 | } |
| 55 | |
| 56 | const ( |
| 57 | graspRetryDelay = int64(5 * 60) |
| 58 | graspSuccessDelay = int64(55 * 60) |
| 59 | ) |
| 60 | |
| 61 | type tickProgress struct { |
| 62 | At int64 `json:"at"` |
| 63 | LastKey string `json:"last_key"` |
| 64 | Repos map[string]tickRepository `json:"repos,omitempty"` |
| 65 | } |
| 66 | |
| 67 | type tickRepository struct { |
| 68 | AttemptedAt int64 `json:"attempted_at"` |
| 69 | SucceededAt int64 `json:"succeeded_at,omitempty"` |
| 70 | NextAt int64 `json:"next_at,omitempty"` |
| 71 | Fingerprint string `json:"fingerprint,omitempty"` |
| 72 | Attempts int `json:"attempts"` |
| 73 | Error string `json:"error,omitempty"` |
| 74 | } |
| 75 | |
| 76 | // RepositorySyncStatus is the durable status of one GRASP reconciliation job. |
| 77 | type RepositorySyncStatus struct { |
| 78 | AttemptedAt int64 `json:"attempted_at"` |
| 79 | SucceededAt int64 `json:"succeeded_at,omitempty"` |
| 80 | NextAt int64 `json:"next_at,omitempty"` |
| 81 | Fingerprint string `json:"fingerprint,omitempty"` |
| 82 | Attempts int `json:"attempts"` |
| 83 | Error string `json:"error,omitempty"` |
| 84 | } |
| 85 | |
| 86 | // SyncStatus is a read-only snapshot suitable for relay status pages. |
| 87 | type SyncStatus struct { |
| 88 | At int64 `json:"at"` |
| 89 | LastKey string `json:"last_key"` |
| 90 | Repositories map[string]RepositorySyncStatus `json:"repositories,omitempty"` |
| 91 | } |
| 92 | |
| 93 | func (g *GitRelay) GRASPService() *Service { return &Service{relay: g} } |
| 94 | |
| 95 | func (s *Service) progressPath() string { |
| 96 | return filepath.Join(s.relay.root, ".tinyrelay" , "grasp-progress.json" ) |
| 97 | } |
| 98 | |
| 99 | func (s *Service) SyncStatus(ctx context.Context) (SyncStatus, error) { |
| 100 | if s == nil || s.relay == nil { |
| 101 | return SyncStatus{}, errors.New("git relay: nil GRASP service" ) |
| 102 | } |
| 103 | select { |
| 104 | case <-ctx.Done(): |
| 105 | return SyncStatus{}, ctx.Err() |
| 106 | default: |
| 107 | } |
| 108 | b, err := os.ReadFile(s.progressPath()) |
| 109 | if errors.Is(err, os.ErrNotExist) { |
| 110 | return SyncStatus{Repositories: make(map[string]RepositorySyncStatus)}, nil |
| 111 | } |
| 112 | if err != nil { |
| 113 | return SyncStatus{}, err |
| 114 | } |
| 115 | var progress tickProgress |
| 116 | if err := json.Unmarshal(b, &progress); err != nil { |
| 117 | return SyncStatus{}, fmt.Errorf("decode GRASP progress: %w" , err) |
| 118 | } |
| 119 | status := SyncStatus{At: progress.At, LastKey: progress.LastKey, Repositories: make(map[string]RepositorySyncStatus, len(progress.Repos))} |
| 120 | for key, repo := range progress.Repos { |
| 121 | status.Repositories[key] = RepositorySyncStatus{AttemptedAt: repo.AttemptedAt, SucceededAt: repo.SucceededAt, NextAt: repo.NextAt, Fingerprint: repo.Fingerprint, Attempts: repo.Attempts, Error: repo.Error} |
| 122 | } |
| 123 | return status, nil |
| 124 | } |
| 125 | |
| 126 | func (s *Service) Tick(ctx context.Context) error { |
| 127 | if s == nil || s.relay == nil { |
| 128 | return errors.New("git relay: nil GRASP service" ) |
| 129 | } |
| 130 | s.relay.tickMu.Lock() |
| 131 | defer s.relay.tickMu.Unlock() |
| 132 | s.relay.mu.RLock() |
| 133 | repos := make([]Repository, 0, len(s.relay.repos)) |
| 134 | for _, r := range s.relay.repos { |
| 135 | repos = append(repos, r) |
| 136 | } |
| 137 | s.relay.mu.RUnlock() |
| 138 | sort.Slice(repos, func(i, j int) bool { |
| 139 | return key(repos[i].Owner, repos[i].Identifier) < key(repos[j].Owner, repos[j].Identifier) |
| 140 | }) |
| 141 | var last string |
| 142 | progress := tickProgress{At: time.Now().Unix(), Repos: make(map[string]tickRepository, len(repos))} |
| 143 | previousProgress, _ := s.loadProgress() |
| 144 | // Resume after the last attempted repository if the prior pass ran out |
| 145 | // of time. Preserve untouched statuses instead of marking them failed. |
| 146 | last = previousProgress.LastKey |
| 147 | for _, r := range repos { |
| 148 | if previous, ok := previousProgress.Repos[key(r.Owner, r.Identifier)]; ok { |
| 149 | progress.Repos[key(r.Owner, r.Identifier)] = previous |
| 150 | } |
| 151 | } |
| 152 | start := sort.Search(len(repos), func(i int) bool { return key(repos[i].Owner, repos[i].Identifier) > last }) |
| 153 | repos = append(repos[start:], repos[:start]...) |
| 154 | var tickErr error |
| 155 | for _, r := range repos { |
| 156 | if ctx.Err() != nil { |
| 157 | tickErr = errors.Join(tickErr, ctx.Err()) |
| 158 | break |
| 159 | } |
| 160 | repoKey := key(r.Owner, r.Identifier) |
| 161 | status := tickRepository{AttemptedAt: progress.At, Attempts: 1} |
| 162 | profile := s.relay.policyValue() |
| 163 | status.Fingerprint = repoFingerprint(r, profile) |
| 164 | if previous, ok := previousProgress.Repos[repoKey]; ok { |
| 165 | if previous.NextAt > progress.At && previous.Fingerprint == status.Fingerprint { |
| 166 | progress.Repos[repoKey] = previous |
| 167 | continue |
| 168 | } |
| 169 | if previous.Error != "" { |
| 170 | status.Attempts = previous.Attempts + 1 |
| 171 | } |
| 172 | } |
| 173 | last = repoKey |
| 174 | features := profile.Features |
| 175 | var repoErr error |
| 176 | if features.Grasp && features.Grasp02 { |
| 177 | if s.relay.eventSync != nil { |
| 178 | eventCtx, cancel := context.WithTimeout(ctx, 40*time.Second) |
| 179 | repoErr = errors.Join(repoErr, s.relay.eventSync(eventCtx, r)) |
| 180 | cancel() |
| 181 | } |
| 182 | gitCtx, cancel := context.WithTimeout(ctx, 20*time.Second) |
| 183 | if s.relay.gitSync != nil { |
| 184 | repoErr = errors.Join(repoErr, s.relay.gitSync(gitCtx, r)) |
| 185 | } else if len(r.Refs) > 0 && len(RepairSources(r)) > 0 { |
| 186 | repoErr = errors.Join(repoErr, s.relay.FetchMissing(gitCtx, r, RepairSources(r), r.Refs)) |
| 187 | } |
| 188 | cancel() |
| 189 | } |
| 190 | if errors.Is(repoErr, ErrIncomplete) { |
| 191 | // Progress is saved; continue soon without recording a failure. |
| 192 | if previous, ok := previousProgress.Repos[repoKey]; ok { |
| 193 | status.SucceededAt = previous.SucceededAt |
| 194 | } |
| 195 | status.NextAt = progress.At + graspRetryDelay |
| 196 | } else if repoErr != nil { |
| 197 | status.Error = repoErr.Error() |
| 198 | status.NextAt = progress.At + graspRetryDelay |
| 199 | tickErr = errors.Join(tickErr, fmt.Errorf("%s synchronization: %w" , last, repoErr)) |
| 200 | } else { |
| 201 | status.SucceededAt = progress.At |
| 202 | status.NextAt = progress.At + graspSuccessDelay |
| 203 | } |
| 204 | progress.Repos[last] = status |
| 205 | } |
| 206 | progress.LastKey = last |
| 207 | b, err := json.Marshal(progress) |
| 208 | if err != nil { |
| 209 | return err |
| 210 | } |
| 211 | if err := os.MkdirAll(filepath.Dir(s.progressPath()), 0700); err != nil { |
| 212 | return err |
| 213 | } |
| 214 | tmp := s.progressPath() + ".tmp" |
| 215 | if err := os.WriteFile(tmp, b, 0600); err != nil { |
| 216 | return err |
| 217 | } |
| 218 | if err := syncFile(tmp); err != nil { |
| 219 | return err |
| 220 | } |
| 221 | if err := os.Rename(tmp, s.progressPath()); err != nil { |
| 222 | return err |
| 223 | } |
| 224 | return tickErr |
| 225 | } |
| 226 | |
| 227 | func repoFingerprint(r Repository, profile Policy) string { |
| 228 | refs := make([]string, 0, len(r.Refs)) |
| 229 | for ref, oid := range r.Refs { |
| 230 | refs = append(refs, ref+"=" +oid) |
| 231 | } |
| 232 | sort.Strings(refs) |
| 233 | profileJSON, _ := json.Marshal(struct { |
| 234 | Grasp bool `json:"grasp"` |
| 235 | Reads string `json:"reads"` |
| 236 | Grasp02 bool `json:"grasp02"` |
| 237 | Grasp03 bool `json:"grasp03"` |
| 238 | Peers []string `json:"peers"` |
| 239 | }{profile.Features.Grasp, profile.Reads, profile.Features.Grasp02, profile.Features.Grasp03, profile.PrivatePeers}) |
| 240 | parts := []string{r.Owner, r.Identifier, r.EventID, r.Head, strconv.FormatBool(r.Private), strings.Join(refs, "\x00" ), strings.Join(r.Clone, "\x00" ), strings.Join(r.Relays, "\x00" ), strings.Join(r.Maintainers, "\x00" ), string(profileJSON)} |
| 241 | sum := sha256.Sum256([]byte(strings.Join(parts, "\x01" ))) |
| 242 | return hex.EncodeToString(sum[:]) |
| 243 | } |
| 244 | |
| 245 | func (s *Service) loadProgress() (tickProgress, error) { |
| 246 | b, err := os.ReadFile(s.progressPath()) |
| 247 | if err != nil { |
| 248 | return tickProgress{}, err |
| 249 | } |
| 250 | var progress tickProgress |
| 251 | if err := json.Unmarshal(b, &progress); err != nil { |
| 252 | return tickProgress{}, err |
| 253 | } |
| 254 | return progress, nil |
| 255 | } |
| 256 | |
| 257 | // RepairSources gives a daemon a deterministic source order for a missing |
| 258 | // object. Clone URLs are preferred, followed by relay URLs; callers must still |
| 259 | // pass each URL through AdmitSource and verify the object graph. |
| 260 | func RepairSources(r Repository) []string { |
| 261 | out := append([]string(nil), r.Clone...) |
| 262 | out = append(out, r.Relays...) |
| 263 | return out |
| 264 | } |