tinyrelayfile
tinyrelay/internal/daemon/custom_views_transform.go/
internal/daemon/custom_views_transform.go raw
1package daemon
2
3// Custom view transforms. Each source event is one POST of its fenced
4// blocks to the view's transform, signed with the view's secret. What comes
5// back is checked, signed by the relay and attached to the source. A failed
6// POST is tried again after one minute and then after five; after 20
7// failures in a row the view is paused until the owner resumes it.
8
9import (
10 "bytes"
11 "context"
12 "crypto/sha256"
13 "crypto/tls"
14 "database/sql"
15 "encoding/base64"
16 "encoding/hex"
17 "encoding/json"
18 "errors"
19 "fmt"
20 "io"
21 "net/http"
22 "sort"
23 "strconv"
24 "strings"
25 "sync"
26 "time"
27
28 "github.com/FelineStateMachine/tinyrelay/internal/event"
29 "github.com/FelineStateMachine/tinyrelay/internal/replication"
30 "github.com/FelineStateMachine/tinyrelay/internal/views"
31 "github.com/FelineStateMachine/tinyrelay/internal/work"
32 gitrelay "github.com/FelineStateMachine/tinyrelay/tinygit"
33)
34
35// viewPayload is the work intent body: the source event, so the run does
36// not depend on the event still being stored, the attempt number and the
37// run that queued it.
38type viewPayload struct {
39 Event event.Event `json:"event"`
40 Attempt int `json:"attempt"`
41 Run string `json:"run,omitempty"`
42 Refresh bool `json:"refresh,omitempty"`
43 Generation string `json:"generation,omitempty"`
44}
45
46// viewRequest is the body posted to a transform: only the blocks, with the
47// source named so the transform can correlate. No event content, tags,
48// author or time leaves the relay.
49type viewRequest struct {
50 Relay string `json:"relay"`
51 View string `json:"view"`
52 Source viewSource `json:"source"`
53 Blocks []views.Block `json:"blocks"`
54}
55
56type viewSource struct {
57 ID string `json:"id"`
58 Kind int `json:"kind"`
59}
60
61// viewResponse is what a transform answers with.
62type viewResponse struct {
63 Artifacts []viewArtifact `json:"artifacts"`
64 Errors []viewError `json:"errors"`
65}
66
67type viewArtifact struct {
68 Block int `json:"block"`
69 Type string `json:"type"`
70 Body string `json:"body"`
71 Engine string `json:"engine"`
72}
73
74type viewError struct {
75 Block int `json:"block"`
76 Error string `json:"error"`
77}
78
79// viewSourceKey names a source in the artifact tables: the event id, or the
80// coordinate of a repository state so a new state replaces its README.
81func viewSourceKey(e event.Event) string {
82 if e.Kind == event.KIND_REPO_STATE {
83 return strconv.Itoa(e.Kind) + ":" + e.PubKey + ":" + event.Tag(e, "d")
84 }
85 return e.ID
86}
87
88// viewSourceTag is the tag an artifact carries for a source.
89func viewSourceTag(source string) []string {
90 if strings.HasPrefix(source, strconv.Itoa(event.KIND_REPO_STATE)+":") {
91 return []string{"a", source}
92 }
93 return []string{"e", source}
94}
95
96// viewHTTPClient refuses private addresses and redirects unless the relay
97// itself runs on a loopback address.
98func (t *customViewService) viewHTTPClient() *http.Client {
99 if t.viewClient != nil {
100 return t.viewClient
101 }
102 if t.loopback != nil && t.loopback() {
103 return &http.Client{Timeout: viewTimeout, Transport: &http.Transport{TLSClientConfig: &tls.Config{MinVersion: tls.VersionTLS12}}, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}
104 }
105 return replication.NewPinnedClient(viewTimeout)
106}
107
108// viewLock bounds work to one transform run per view at a time.
109func (t *customViewService) viewLock(name string) *sync.Mutex {
110 t.viewMu.Lock()
111 defer t.viewMu.Unlock()
112 if t.viewLocks == nil {
113 t.viewLocks = map[string]*sync.Mutex{}
114 }
115 lock := t.viewLocks[name]
116 if lock == nil {
117 lock = &sync.Mutex{}
118 t.viewLocks[name] = lock
119 }
120 return lock
121}
122
123// sourceBlocks finds the blocks of a source: the fenced blocks of its
124// content, or of the README at the head of a repository state.
125func (t *customViewService) sourceBlocks(ctx context.Context, e event.Event) ([]views.Block, error) {
126 if e.Kind != event.KIND_REPO_STATE {
127 return views.Blocks(e.Content), nil
128 }
129 readme, err := t.stateReadme(ctx, e)
130 if err != nil {
131 return nil, err
132 }
133 return views.Blocks(readme), nil
134}
135
136// stateReadme reads README.md at the head named by a repository state. A
137// missing README yields no content.
138func (t *customViewService) stateReadme(ctx context.Context, e event.Event) (string, error) {
139 if t.git == nil {
140 return "", nil
141 }
142 head := strings.TrimPrefix(event.Tag(e, "HEAD"), "ref: ")
143 commit := ""
144 for _, tag := range e.Tags {
145 if len(tag) >= 2 && tag[0] == head {
146 commit = tag[1]
147 }
148 }
149 if head == "" || commit == "" {
150 return "", nil
151 }
152 for _, name := range []string{"README.md", "readme.md", "README"} {
153 page, err := t.git.Browse(ctx, gitrelay.BrowseRequest{Owner: e.PubKey, Repo: event.Tag(e, "d"), Ref: commit, Path: name, View: "file", Limit: 1})
154 if err != nil {
155 if strings.HasPrefix(err.Error(), "not found:") || strings.HasPrefix(err.Error(), "invalid:") {
156 continue
157 }
158 return "", err
159 }
160 if page.Binary || page.Content == "" {
161 continue
162 }
163 return page.Content, nil
164 }
165 return "", nil
166}
167
168// sourceCurrent reports whether the source is still the stored event: an
169// event that was deleted, or a state that a newer state replaced, renders
170// nothing.
171func (t *customViewService) sourceCurrent(ctx context.Context, e event.Event) (bool, error) {
172 var id string
173 var err error
174 if e.Kind == event.KIND_REPO_STATE {
175 err = t.store.DB().QueryRowContext(ctx, `SELECT id FROM events WHERE kind=? AND pubkey=? AND d=? AND (expires=0 OR expires>?) ORDER BY created_at DESC,id ASC LIMIT 1`, e.Kind, e.PubKey, event.Tag(e, "d"), time.Now().Unix()).Scan(&id)
176 } else {
177 err = t.store.DB().QueryRowContext(ctx, `SELECT id FROM events WHERE id=? AND (expires=0 OR expires>?)`, e.ID, time.Now().Unix()).Scan(&id)
178 }
179 if errors.Is(err, sql.ErrNoRows) {
180 return false, nil
181 }
182 return id == e.ID, err
183}
184
185// handleViewTransform serves one view-transform intent. It rechecks the
186// view and the source right before the POST, attaches artifacts that
187// already exist for a block's hash and posts only the blocks that need
188// rendering.
189func (t *customViewService) handleViewTransform(ctx context.Context, intent work.Intent) (err error) {
190 ctx, finish := t.telemetry.Start(ctx, "view")
191 outcome := "error"
192 defer func() { finish(outcome) }()
193 var payload viewPayload
194 if err := json.Unmarshal([]byte(intent.Payload), &payload); err != nil || payload.Event.ID == "" {
195 outcome = "invalid"
196 return errors.New("view-transform: invalid payload")
197 }
198 if payload.Attempt < 1 {
199 payload.Attempt = 1
200 }
201 lock := t.viewLock(intent.Target)
202 lock.Lock()
203 defer lock.Unlock()
204 view, err := t.customViewByName(ctx, intent.Target)
205 if err != nil {
206 if strings.HasPrefix(err.Error(), "not found:") {
207 outcome = "paused"
208 return nil
209 }
210 return err
211 }
212 if !view.Enabled {
213 outcome = "paused"
214 return nil
215 }
216 if payload.Generation != "" && payload.Generation != view.generation {
217 outcome = "invalid"
218 return nil
219 }
220 current, err := t.sourceCurrent(ctx, payload.Event)
221 if err != nil {
222 return err
223 }
224 if !current {
225 outcome = "invalid"
226 return nil
227 }
228 blocks, err := t.sourceBlocks(ctx, payload.Event)
229 if err != nil {
230 return err
231 }
232 blocks = views.Matching(blocks, view.Languages)
233 source := viewSourceKey(payload.Event)
234 expires := event.Expiration(payload.Event)
235 // A rebuild deliberately bypasses existing artifacts. The replacement is
236 // written only after the transform succeeds, so a failed rebuild preserves
237 // the previous artifact.
238 var pending []views.Block
239 hashes := make([]string, 0, len(blocks))
240 for _, block := range blocks {
241 hash := views.Hash(block.Lang, block.Source)
242 hashes = append(hashes, hash)
243 if payload.Refresh {
244 pending = append(pending, block)
245 continue
246 }
247 exists, err := t.attachArtifact(ctx, view, hash, source, block.Index, expires)
248 if err != nil {
249 return err
250 }
251 if !exists {
252 pending = append(pending, block)
253 }
254 }
255 if err := t.detachStaleArtifacts(ctx, view.Name, source, hashes); err != nil {
256 return err
257 }
258 if len(pending) == 0 {
259 outcome = "ok"
260 return t.pruneViewArtifacts(ctx, view.Name, intent.ID)
261 }
262 status, response, postErr := t.postViewTransform(ctx, view, payload.Event, pending)
263 now := time.Now().Unix()
264 if postErr == nil {
265 stored, refused := 0, 0
266 sent := map[int]views.Block{}
267 storedBlocks := map[int]bool{}
268 for _, block := range pending {
269 sent[block.Index] = block
270 }
271 for _, artifact := range response.Artifacts {
272 block, ok := sent[artifact.Block]
273 if !ok {
274 refused++
275 continue
276 }
277 if storedBlocks[artifact.Block] {
278 refused++
279 continue
280 }
281 if err := t.storeArtifact(ctx, view, block, artifact, source, expires, now); err != nil {
282 if strings.HasPrefix(err.Error(), "invalid:") {
283 refused++
284 continue
285 }
286 return err
287 }
288 storedBlocks[artifact.Block] = true
289 stored++
290 }
291 if stored > 0 {
292 outcome = "ok"
293 if stored < len(pending) {
294 outcome = "partial"
295 }
296 if _, err := t.store.DB().ExecContext(ctx, `UPDATE custom_views SET last_run_at=?, last_status=?, failures=0 WHERE name=?`, now, outcome, view.Name); err != nil {
297 return err
298 }
299 // Counts only: no source, block or body leaves the relay in a log.
300 t.telemetry.Logger().Info("view transform delivered", "tenant", t.tenantName, "attempt", payload.Attempt, "status", status, "blocks", len(pending), "artifacts", stored, "refused", refused, "errors", len(response.Errors))
301 return t.pruneViewArtifacts(ctx, view.Name, intent.ID)
302 }
303 postErr = errors.New("no valid artifacts")
304 }
305 reason := postErr.Error()
306 failures := view.Failures + 1
307 paused := failures >= viewPauseFailures
308 if paused {
309 outcome = "paused"
310 if err := t.pauseCustomView(ctx, view.Name, failures, fmt.Sprintf("paused after %d failures: %s", failures, reason)); err != nil {
311 return err
312 }
313 } else if _, err := t.store.DB().ExecContext(ctx, `UPDATE custom_views SET last_run_at=?, failures=?, last_status=? WHERE name=?`, now, failures, reason, view.Name); err != nil {
314 return err
315 }
316 t.telemetry.Logger().Info("view transform failed", "tenant", t.tenantName, "attempt", payload.Attempt, "status", status, "failures", failures, "paused", paused)
317 if paused || payload.Attempt >= viewMaxAttempts {
318 return nil
319 }
320 return t.retryViewTransform(ctx, intent, payload)
321}
322
323// pauseCustomView stops the view and records the count and reason, then
324// drops the index so no new intents are queued for it.
325func (t *customViewService) pauseCustomView(ctx context.Context, name string, failures int, status string) error {
326 if _, err := t.store.DB().ExecContext(ctx, `UPDATE custom_views SET enabled=0, failures=?, last_status=? WHERE name=?`, failures, status, name); err != nil {
327 return err
328 }
329 t.invalidateCustomViews()
330 return nil
331}
332
333// retryViewTransform queues the next attempt after its backoff as its own
334// intent so the durable queue's ordering and fencing still apply.
335func (t *customViewService) retryViewTransform(ctx context.Context, intent work.Intent, payload viewPayload) error {
336 delay := viewBackoff[len(viewBackoff)-1]
337 if payload.Attempt-1 < len(viewBackoff) {
338 delay = viewBackoff[payload.Attempt-1]
339 }
340 payload.Attempt++
341 encoded, err := json.Marshal(payload)
342 if err != nil {
343 return err
344 }
345 now := time.Now().Unix()
346 // The intent id names the source and run; earlier attempts add "#n".
347 source, _, _ := strings.Cut(intent.EventID, "#")
348 key := sha256.Sum256([]byte(viewTransform + "\x00" + source + "\x00" + intent.Target + "\x00" + fmt.Sprint(payload.Attempt)))
349 return t.store.WithTx(ctx, func(tx *sql.Tx) error {
350 _, err := tx.ExecContext(ctx, `INSERT OR IGNORE INTO work_intents(id,kind,event_id,target,payload,next_at,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?)`, hex.EncodeToString(key[:]), viewTransform, source+"#"+fmt.Sprint(payload.Attempt), intent.Target, string(encoded), now+int64(delay/time.Second), now, now)
351 return err
352 })
353}
354
355// postViewTransform sends the blocks and reads the answer. It returns the
356// HTTP status with a bounded reason on failure; a response that is not
357// JSON counts as a failure.
358func (t *customViewService) postViewTransform(ctx context.Context, view customView, e event.Event, blocks []views.Block) (int, viewResponse, error) {
359 body, err := json.Marshal(viewRequest{Relay: t.publicURL, View: view.Name, Source: viewSource{ID: viewSourceKey(e), Kind: e.Kind}, Blocks: blocks})
360 if err != nil {
361 return 0, viewResponse{}, errors.New("encode request")
362 }
363 ctx, cancel := context.WithTimeout(ctx, viewTimeout)
364 defer cancel()
365 request, err := http.NewRequestWithContext(ctx, http.MethodPost, view.Transform, bytes.NewReader(body))
366 if err != nil {
367 return 0, viewResponse{}, errors.New("bad url")
368 }
369 request.Header.Set("Content-Type", "application/json")
370 request.Header.Set("Accept", "application/json")
371 request.Header.Set("X-Tiny-View", view.Name)
372 request.Header.Set("X-Tiny-Relay", t.publicURL)
373 signature := callbackSignature(view.secret, body)
374 request.Header.Set("X-Tiny-Signature", signature)
375 request.Header.Set("X-Transform-View", view.Name)
376 request.Header.Set("X-Transform-Relay", t.publicURL)
377 request.Header.Set("X-Transform-Signature", signature)
378 request.Header.Set("User-Agent", "tinyrelay")
379 response, err := t.viewHTTPClient().Do(request)
380 if err != nil {
381 if errors.Is(err, context.DeadlineExceeded) || ctx.Err() != nil {
382 return 0, viewResponse{}, errors.New("timeout")
383 }
384 return 0, viewResponse{}, errors.New("connection failed")
385 }
386 defer response.Body.Close()
387 raw, err := io.ReadAll(io.LimitReader(response.Body, viewResponseMaxBytes))
388 if err != nil {
389 return response.StatusCode, viewResponse{}, errors.New("read failed")
390 }
391 if response.StatusCode < 200 || response.StatusCode >= 300 {
392 return response.StatusCode, viewResponse{}, fmt.Errorf("HTTP %d", response.StatusCode)
393 }
394 var decoded viewResponse
395 if err := json.Unmarshal(raw, &decoded); err != nil {
396 return response.StatusCode, viewResponse{}, errors.New("invalid response")
397 }
398 return response.StatusCode, decoded, nil
399}
400
401// storeArtifact checks one artifact and keeps it as a signed record
402// attached to its source.
403func (t *customViewService) storeArtifact(ctx context.Context, view customView, block views.Block, artifact viewArtifact, source string, expires int64, now int64) error {
404 content, err := checkArtifact(artifact, view.MaxBytes)
405 if err != nil {
406 return err
407 }
408 engine := strings.TrimSpace(artifact.Engine)
409 if len(engine) > 64 {
410 engine = engine[:64]
411 }
412 hash := views.Hash(block.Lang, block.Source)
413 err = t.store.WithTx(ctx, func(tx *sql.Tx) error {
414 if _, err := tx.ExecContext(ctx, `INSERT INTO custom_view_artifacts(view,hash,event_id,type,engine,audience,content,raw,created_at) VALUES(?,?,'',?,?,?,?,'',?) ON CONFLICT(view,hash) DO UPDATE SET type=excluded.type, engine=excluded.engine, audience=excluded.audience, content=excluded.content, created_at=excluded.created_at`, view.Name, hash, artifact.Type, engine, view.Audience, content, now); err != nil {
415 return err
416 }
417 _, err := tx.ExecContext(ctx, `INSERT INTO custom_view_sources(view,hash,source,block,expires) VALUES(?,?,?,?,?) ON CONFLICT(view,hash,source) DO UPDATE SET block=excluded.block, expires=excluded.expires`, view.Name, hash, source, block.Index, expires)
418 return err
419 })
420 if err != nil {
421 return err
422 }
423 return t.signArtifact(ctx, view, hash, now)
424}
425
426// attachArtifact reports whether an artifact exists for the hash and, when
427// it does, attaches the source to it through a fresh signed record.
428func (t *customViewService) attachArtifact(ctx context.Context, view customView, hash, source string, block int, expires int64) (bool, error) {
429 var exists int
430 err := t.store.DB().QueryRowContext(ctx, `SELECT 1 FROM custom_view_artifacts WHERE view=? AND hash=?`, view.Name, hash).Scan(&exists)
431 if errors.Is(err, sql.ErrNoRows) {
432 return false, nil
433 }
434 if err != nil {
435 return false, err
436 }
437 var current int
438 var currentBlock int
439 var currentExpires int64
440 err = t.store.DB().QueryRowContext(ctx, `SELECT 1,block,expires FROM custom_view_sources WHERE view=? AND hash=? AND source=?`, view.Name, hash, source).Scan(&current, &currentBlock, &currentExpires)
441 if err == nil && currentBlock == block && currentExpires == expires {
442 return true, nil
443 }
444 if err != nil && !errors.Is(err, sql.ErrNoRows) {
445 return false, err
446 }
447 if _, err := t.store.DB().ExecContext(ctx, `INSERT INTO custom_view_sources(view,hash,source,block,expires) VALUES(?,?,?,?,?) ON CONFLICT(view,hash,source) DO UPDATE SET block=excluded.block, expires=excluded.expires`, view.Name, hash, source, block, expires); err != nil {
448 return false, err
449 }
450 return true, t.signArtifact(ctx, view, hash, time.Now().Unix())
451}
452
453// signArtifact signs the artifact with its current sources as a kind 30078
454// record and stores the record when the view is public.
455func (t *customViewService) signArtifact(ctx context.Context, view customView, hash string, now int64) error {
456 var typ, engine, content string
457 if err := t.store.DB().QueryRowContext(ctx, `SELECT type,engine,content FROM custom_view_artifacts WHERE view=? AND hash=?`, view.Name, hash).Scan(&typ, &engine, &content); err != nil {
458 return err
459 }
460 rows, err := t.store.DB().QueryContext(ctx, `SELECT source,block,expires FROM custom_view_sources WHERE view=? AND hash=? ORDER BY rowid`, view.Name, hash)
461 if err != nil {
462 return err
463 }
464 tags := [][]string{{"-"}, {"d", "bind.ws/view/" + view.Name + "/" + hash}, {"view", view.Name}}
465 var blocks [][]string
466 expires := int64(0)
467 unlimited := false
468 for rows.Next() {
469 var source string
470 var block int
471 var sourceExpires int64
472 if err := rows.Scan(&source, &block, &sourceExpires); err != nil {
473 rows.Close()
474 return err
475 }
476 tags = append(tags, viewSourceTag(source))
477 blocks = append(blocks, []string{"block", strconv.Itoa(block)})
478 if sourceExpires == 0 {
479 unlimited = true
480 } else if sourceExpires > expires {
481 expires = sourceExpires
482 }
483 }
484 if err := errors.Join(rows.Err(), rows.Close()); err != nil {
485 return err
486 }
487 tags = append(tags, blocks...)
488 tags = append(tags, []string{"type", typ})
489 if engine != "" {
490 tags = append(tags, []string{"engine", engine})
491 }
492 if expires > 0 && !unlimited {
493 tags = append(tags, []string{"expiration", strconv.FormatInt(expires, 10)})
494 }
495 var signed event.Event
496 if view.Audience == "public" {
497 signed, err = t.records.Generate(ctx, event.KIND_VIEW, tags, content, now)
498 } else {
499 signed, err = t.records.Sign(ctx, event.KIND_VIEW, tags, content, now)
500 }
501 if err != nil {
502 return err
503 }
504 raw, err := json.Marshal(signed)
505 if err != nil {
506 return err
507 }
508 _, err = t.store.DB().ExecContext(ctx, `UPDATE custom_view_artifacts SET event_id=?, raw=? WHERE view=? AND hash=?`, signed.ID, string(raw), view.Name, hash)
509 return err
510}
511
512// detachStaleArtifacts drops the source from artifacts whose block it no
513// longer carries, then removes artifacts left without a source.
514func (t *customViewService) detachStaleArtifacts(ctx context.Context, name, source string, hashes []string) error {
515 rows, err := t.store.DB().QueryContext(ctx, `SELECT hash FROM custom_view_sources WHERE view=? AND source=?`, name, source)
516 if err != nil {
517 return err
518 }
519 keep := map[string]bool{}
520 for _, hash := range hashes {
521 keep[hash] = true
522 }
523 var stale []string
524 for rows.Next() {
525 var hash string
526 if err := rows.Scan(&hash); err != nil {
527 rows.Close()
528 return err
529 }
530 if !keep[hash] {
531 stale = append(stale, hash)
532 }
533 }
534 if err := errors.Join(rows.Err(), rows.Close()); err != nil {
535 return err
536 }
537 for _, hash := range stale {
538 if _, err := t.store.DB().ExecContext(ctx, `DELETE FROM custom_view_sources WHERE view=? AND hash=? AND source=?`, name, hash, source); err != nil {
539 return err
540 }
541 }
542 return t.dropOrphanArtifacts(ctx, name)
543}
544
545// pruneViewArtifacts detaches sources that no longer exist and removes the
546// artifacts that lose their last source. A view with transform work still
547// queued is left alone so a replaced source can reattach its blocks first.
548func (t *customViewService) pruneViewArtifacts(ctx context.Context, name, except string) error {
549 var pending int
550 if err := t.store.DB().QueryRowContext(ctx, `SELECT count(*) FROM work_intents WHERE kind=? AND target=? AND state IN ('pending','running') AND id<>?`, viewTransform, name, except).Scan(&pending); err != nil {
551 return err
552 }
553 if pending > 0 {
554 return nil
555 }
556 rows, err := t.store.DB().QueryContext(ctx, `SELECT DISTINCT source FROM custom_view_sources WHERE view=?`, name)
557 if err != nil {
558 return err
559 }
560 var sources []string
561 for rows.Next() {
562 var source string
563 if err := rows.Scan(&source); err != nil {
564 rows.Close()
565 return err
566 }
567 sources = append(sources, source)
568 }
569 if err := errors.Join(rows.Err(), rows.Close()); err != nil {
570 return err
571 }
572 removed := 0
573 for _, source := range sources {
574 present, err := t.sourcePresent(ctx, source)
575 if err != nil {
576 return err
577 }
578 if present {
579 continue
580 }
581 if _, err := t.store.DB().ExecContext(ctx, `DELETE FROM custom_view_sources WHERE view=? AND source=?`, name, source); err != nil {
582 return err
583 }
584 removed++
585 }
586 if removed == 0 {
587 return nil
588 }
589 return t.dropOrphanArtifacts(ctx, name)
590}
591
592// pruneAllViewArtifacts runs the prune for every view; the maintenance
593// sweep calls it after expired events are removed.
594func (t *customViewService) pruneAllViewArtifacts(ctx context.Context) error {
595 rows, err := t.customViewRows(ctx)
596 if err != nil {
597 return err
598 }
599 for _, view := range rows {
600 if err := t.pruneViewArtifacts(ctx, view.Name, ""); err != nil {
601 return err
602 }
603 }
604 return nil
605}
606
607func (t *customViewService) sourcePresent(ctx context.Context, source string) (bool, error) {
608 var one int
609 var err error
610 if tag := viewSourceTag(source); tag[0] == "a" {
611 parts := strings.SplitN(source, ":", 3)
612 if len(parts) != 3 {
613 return false, nil
614 }
615 err = t.store.DB().QueryRowContext(ctx, `SELECT 1 FROM events WHERE kind=? AND pubkey=? AND d=? AND (expires=0 OR expires>?) LIMIT 1`, event.KIND_REPO_STATE, parts[1], parts[2], time.Now().Unix()).Scan(&one)
616 } else {
617 err = t.store.DB().QueryRowContext(ctx, `SELECT 1 FROM events WHERE id=? AND (expires=0 OR expires>?) LIMIT 1`, source, time.Now().Unix()).Scan(&one)
618 }
619 if errors.Is(err, sql.ErrNoRows) {
620 return false, nil
621 }
622 return err == nil, err
623}
624
625// dropOrphanArtifacts removes artifacts without a source, and their stored
626// records, then re-signs the artifacts whose source list shrank.
627func (t *customViewService) dropOrphanArtifacts(ctx context.Context, name string) error {
628 ids, err := t.customViewArtifactIDs(ctx, `SELECT event_id FROM custom_view_artifacts WHERE view=? AND NOT EXISTS(SELECT 1 FROM custom_view_sources WHERE custom_view_sources.view=custom_view_artifacts.view AND custom_view_sources.hash=custom_view_artifacts.hash)`, name)
629 if err != nil {
630 return err
631 }
632 if _, err := t.store.DB().ExecContext(ctx, `DELETE FROM custom_view_artifacts WHERE view=? AND NOT EXISTS(SELECT 1 FROM custom_view_sources WHERE custom_view_sources.view=custom_view_artifacts.view AND custom_view_sources.hash=custom_view_artifacts.hash)`, name); err != nil {
633 return err
634 }
635 if err := t.deleteStoredArtifacts(ctx, ids); err != nil {
636 return err
637 }
638 if len(ids) > 0 {
639 t.telemetry.Logger().Info("view artifacts removed", "tenant", t.tenantName, "removed", len(ids))
640 }
641 // Artifacts that kept a source but lost one carry a tag for it; a fresh
642 // record drops it.
643 view, err := t.customViewByName(ctx, name)
644 if err != nil {
645 return nil
646 }
647 rows, err := t.store.DB().QueryContext(ctx, `SELECT hash, raw FROM custom_view_artifacts WHERE view=?`, name)
648 if err != nil {
649 return err
650 }
651 // Read the rows out before asking about sources: the store lends one
652 // connection at a time.
653 tagged := map[string][]string{}
654 for rows.Next() {
655 var hash, raw string
656 if err := rows.Scan(&hash, &raw); err != nil {
657 rows.Close()
658 return err
659 }
660 var signed event.Event
661 if json.Unmarshal([]byte(raw), &signed) != nil {
662 continue
663 }
664 sources := []string{}
665 for _, tag := range signed.Tags {
666 if len(tag) == 2 && (tag[0] == "e" || tag[0] == "a") {
667 sources = append(sources, tag[1])
668 }
669 }
670 tagged[hash] = sources
671 }
672 if err := errors.Join(rows.Err(), rows.Close()); err != nil {
673 return err
674 }
675 for hash, sources := range tagged {
676 current, err := t.artifactSources(ctx, name, hash)
677 if err != nil {
678 return err
679 }
680 sort.Strings(sources)
681 sort.Strings(current)
682 if strings.Join(sources, "\n") == strings.Join(current, "\n") {
683 continue
684 }
685 if err := t.signArtifact(ctx, view, hash, time.Now().Unix()); err != nil {
686 return err
687 }
688 }
689 return nil
690}
691
692func (t *customViewService) artifactSources(ctx context.Context, name, hash string) ([]string, error) {
693 rows, err := t.store.DB().QueryContext(ctx, `SELECT source FROM custom_view_sources WHERE view=? AND hash=?`, name, hash)
694 if err != nil {
695 return nil, err
696 }
697 defer rows.Close()
698 var out []string
699 for rows.Next() {
700 var source string
701 if err := rows.Scan(&source); err != nil {
702 return nil, err
703 }
704 out = append(out, source)
705 }
706 return out, rows.Err()
707}
708
709func (t *customViewService) customViewArtifactIDs(ctx context.Context, query string, args ...any) ([]string, error) {
710 rows, err := t.store.DB().QueryContext(ctx, query, args...)
711 if err != nil {
712 return nil, err
713 }
714 defer rows.Close()
715 var ids []string
716 for rows.Next() {
717 var id string
718 if err := rows.Scan(&id); err != nil {
719 return nil, err
720 }
721 if id != "" {
722 ids = append(ids, id)
723 }
724 }
725 return ids, rows.Err()
726}
727
728// deleteStoredArtifacts removes the public records of artifacts by id.
729func (t *customViewService) deleteStoredArtifacts(ctx context.Context, ids []string) error {
730 for _, id := range ids {
731 if _, err := t.store.DeleteEvent(ctx, id); err != nil {
732 return err
733 }
734 }
735 return nil
736}
737
738// cascadeViewDeletion runs after a kind 5 deletion is stored: artifacts
739// attached to the deleted events or addresses lose that source at once.
740func (t *customViewService) cascadeViewDeletion(ctx context.Context, e event.Event) {
741 if e.Kind != event.KIND_DELETION {
742 return
743 }
744 rows, err := t.customViewRows(ctx)
745 if err != nil || len(rows) == 0 {
746 return
747 }
748 for _, view := range rows {
749 if err := t.pruneViewArtifacts(ctx, view.Name, ""); err != nil {
750 t.telemetry.Logger().Error("view artifacts not pruned", "tenant", t.tenantName, "error", err)
751 }
752 }
753}
754
755// artifactRecord is one stored artifact as the route serves it.
756type artifactRecord struct {
757 Type string
758 Audience string
759 Content string
760 EventID string
761}
762
763func (t *customViewService) customViewArtifact(ctx context.Context, name, hash string) (artifactRecord, error) {
764 var record artifactRecord
765 err := t.store.DB().QueryRowContext(ctx, `SELECT type,audience,content,event_id FROM custom_view_artifacts WHERE view=? AND hash=?`, name, hash).Scan(&record.Type, &record.Audience, &record.Content, &record.EventID)
766 if errors.Is(err, sql.ErrNoRows) {
767 return artifactRecord{}, errors.New("not found: artifact")
768 }
769 return record, err
770}
771
772// body returns the bytes served for the artifact.
773func (r artifactRecord) body() ([]byte, error) {
774 if r.Type == "image/png" {
775 return base64.StdEncoding.DecodeString(r.Content)
776 }
777 return []byte(r.Content), nil
778}

About

ownerc01503e8035f
visibilitypublic
headd730679639bf

Maintainers

  • c01503e8035f owner
  • ba1b5beed5b5 agent hermes

Clone

  • https://012.run/npub1cq2s86qrtadf36eqq3epm38wmhdynrrcwxf84eldk3f94dkv7ttsrulrz5/tinyrelay.git

Refs