1package daemon23// Custom view transforms. Each source event is one POST of its fenced4// blocks to the view's transform, signed with the view's secret. What comes5// back is checked, signed by the relay and attached to the source. A failed6// POST is tried again after one minute and then after five; after 207// failures in a row the view is paused until the owner resumes it.89import (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"2728 "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)3435// viewPayload is the work intent body: the source event, so the run does36// not depend on the event still being stored, the attempt number and the37// 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}4546// viewRequest is the body posted to a transform: only the blocks, with the47// 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}5556type viewSource struct {57 ID string `json:"id"`58 Kind int `json:"kind"`59}6061// viewResponse is what a transform answers with.62type viewResponse struct {63 Artifacts []viewArtifact `json:"artifacts"`64 Errors []viewError `json:"errors"`65}6667type viewArtifact struct {68 Block int `json:"block"`69 Type string `json:"type"`70 Body string `json:"body"`71 Engine string `json:"engine"`72}7374type viewError struct {75 Block int `json:"block"`76 Error string `json:"error"`77}7879// viewSourceKey names a source in the artifact tables: the event id, or the80// 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.ID86}8788// 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}9596// viewHTTPClient refuses private addresses and redirects unless the relay97// itself runs on a loopback address.98func (t *customViewService) viewHTTPClient() *http.Client {99 if t.viewClient != nil {100 return t.viewClient101 }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}107108// 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] = lock119 }120 return lock121}122123// sourceBlocks finds the blocks of a source: the fenced blocks of its124// 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), nil128 }129 readme, err := t.stateReadme(ctx, e)130 if err != nil {131 return nil, err132 }133 return views.Blocks(readme), nil134}135136// stateReadme reads README.md at the head named by a repository state. A137// missing README yields no content.138func (t *customViewService) stateReadme(ctx context.Context, e event.Event) (string, error) {139 if t.git == nil {140 return "", nil141 }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 "", nil151 }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 continue157 }158 return "", err159 }160 if page.Binary || page.Content == "" {161 continue162 }163 return page.Content, nil164 }165 return "", nil166}167168// sourceCurrent reports whether the source is still the stored event: an169// event that was deleted, or a state that a newer state replaced, renders170// nothing.171func (t *customViewService) sourceCurrent(ctx context.Context, e event.Event) (bool, error) {172 var id string173 var err error174 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, nil181 }182 return id == e.ID, err183}184185// handleViewTransform serves one view-transform intent. It rechecks the186// view and the source right before the POST, attaches artifacts that187// already exist for a block's hash and posts only the blocks that need188// 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 viewPayload194 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 = 1200 }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 nil209 }210 return err211 }212 if !view.Enabled {213 outcome = "paused"214 return nil215 }216 if payload.Generation != "" && payload.Generation != view.generation {217 outcome = "invalid"218 return nil219 }220 current, err := t.sourceCurrent(ctx, payload.Event)221 if err != nil {222 return err223 }224 if !current {225 outcome = "invalid"226 return nil227 }228 blocks, err := t.sourceBlocks(ctx, payload.Event)229 if err != nil {230 return err231 }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 is236 // written only after the transform succeeds, so a failed rebuild preserves237 // the previous artifact.238 var pending []views.Block239 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 continue246 }247 exists, err := t.attachArtifact(ctx, view, hash, source, block.Index, expires)248 if err != nil {249 return err250 }251 if !exists {252 pending = append(pending, block)253 }254 }255 if err := t.detachStaleArtifacts(ctx, view.Name, source, hashes); err != nil {256 return err257 }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, 0266 sent := map[int]views.Block{}267 storedBlocks := map[int]bool{}268 for _, block := range pending {269 sent[block.Index] = block270 }271 for _, artifact := range response.Artifacts {272 block, ok := sent[artifact.Block]273 if !ok {274 refused++275 continue276 }277 if storedBlocks[artifact.Block] {278 refused++279 continue280 }281 if err := t.storeArtifact(ctx, view, block, artifact, source, expires, now); err != nil {282 if strings.HasPrefix(err.Error(), "invalid:") {283 refused++284 continue285 }286 return err287 }288 storedBlocks[artifact.Block] = true289 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 err298 }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 + 1307 paused := failures >= viewPauseFailures308 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 err312 }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 err315 }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 nil319 }320 return t.retryViewTransform(ctx, intent, payload)321}322323// pauseCustomView stops the view and records the count and reason, then324// 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 err328 }329 t.invalidateCustomViews()330 return nil331}332333// retryViewTransform queues the next attempt after its backoff as its own334// 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 err344 }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 err352 })353}354355// postViewTransform sends the blocks and reads the answer. It returns the356// HTTP status with a bounded reason on failure; a response that is not357// 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 viewResponse395 if err := json.Unmarshal(raw, &decoded); err != nil {396 return response.StatusCode, viewResponse{}, errors.New("invalid response")397 }398 return response.StatusCode, decoded, nil399}400401// storeArtifact checks one artifact and keeps it as a signed record402// 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 err407 }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 err416 }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 err419 })420 if err != nil {421 return err422 }423 return t.signArtifact(ctx, view, hash, now)424}425426// attachArtifact reports whether an artifact exists for the hash and, when427// 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 int430 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, nil433 }434 if err != nil {435 return false, err436 }437 var current int438 var currentBlock int439 var currentExpires int64440 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(¤t, ¤tBlock, ¤tExpires)441 if err == nil && currentBlock == block && currentExpires == expires {442 return true, nil443 }444 if err != nil && !errors.Is(err, sql.ErrNoRows) {445 return false, err446 }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, err449 }450 return true, t.signArtifact(ctx, view, hash, time.Now().Unix())451}452453// signArtifact signs the artifact with its current sources as a kind 30078454// 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 string457 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 err459 }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 err463 }464 tags := [][]string{{"-"}, {"d", "bind.ws/view/" + view.Name + "/" + hash}, {"view", view.Name}}465 var blocks [][]string466 expires := int64(0)467 unlimited := false468 for rows.Next() {469 var source string470 var block int471 var sourceExpires int64472 if err := rows.Scan(&source, &block, &sourceExpires); err != nil {473 rows.Close()474 return err475 }476 tags = append(tags, viewSourceTag(source))477 blocks = append(blocks, []string{"block", strconv.Itoa(block)})478 if sourceExpires == 0 {479 unlimited = true480 } else if sourceExpires > expires {481 expires = sourceExpires482 }483 }484 if err := errors.Join(rows.Err(), rows.Close()); err != nil {485 return err486 }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.Event496 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 err503 }504 raw, err := json.Marshal(signed)505 if err != nil {506 return err507 }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 err510}511512// detachStaleArtifacts drops the source from artifacts whose block it no513// 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 err518 }519 keep := map[string]bool{}520 for _, hash := range hashes {521 keep[hash] = true522 }523 var stale []string524 for rows.Next() {525 var hash string526 if err := rows.Scan(&hash); err != nil {527 rows.Close()528 return err529 }530 if !keep[hash] {531 stale = append(stale, hash)532 }533 }534 if err := errors.Join(rows.Err(), rows.Close()); err != nil {535 return err536 }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 err540 }541 }542 return t.dropOrphanArtifacts(ctx, name)543}544545// pruneViewArtifacts detaches sources that no longer exist and removes the546// artifacts that lose their last source. A view with transform work still547// 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 int550 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 err552 }553 if pending > 0 {554 return nil555 }556 rows, err := t.store.DB().QueryContext(ctx, `SELECT DISTINCT source FROM custom_view_sources WHERE view=?`, name)557 if err != nil {558 return err559 }560 var sources []string561 for rows.Next() {562 var source string563 if err := rows.Scan(&source); err != nil {564 rows.Close()565 return err566 }567 sources = append(sources, source)568 }569 if err := errors.Join(rows.Err(), rows.Close()); err != nil {570 return err571 }572 removed := 0573 for _, source := range sources {574 present, err := t.sourcePresent(ctx, source)575 if err != nil {576 return err577 }578 if present {579 continue580 }581 if _, err := t.store.DB().ExecContext(ctx, `DELETE FROM custom_view_sources WHERE view=? AND source=?`, name, source); err != nil {582 return err583 }584 removed++585 }586 if removed == 0 {587 return nil588 }589 return t.dropOrphanArtifacts(ctx, name)590}591592// pruneAllViewArtifacts runs the prune for every view; the maintenance593// 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 err598 }599 for _, view := range rows {600 if err := t.pruneViewArtifacts(ctx, view.Name, ""); err != nil {601 return err602 }603 }604 return nil605}606607func (t *customViewService) sourcePresent(ctx context.Context, source string) (bool, error) {608 var one int609 var err error610 if tag := viewSourceTag(source); tag[0] == "a" {611 parts := strings.SplitN(source, ":", 3)612 if len(parts) != 3 {613 return false, nil614 }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, nil621 }622 return err == nil, err623}624625// dropOrphanArtifacts removes artifacts without a source, and their stored626// 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 err631 }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 err634 }635 if err := t.deleteStoredArtifacts(ctx, ids); err != nil {636 return err637 }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 fresh642 // record drops it.643 view, err := t.customViewByName(ctx, name)644 if err != nil {645 return nil646 }647 rows, err := t.store.DB().QueryContext(ctx, `SELECT hash, raw FROM custom_view_artifacts WHERE view=?`, name)648 if err != nil {649 return err650 }651 // Read the rows out before asking about sources: the store lends one652 // connection at a time.653 tagged := map[string][]string{}654 for rows.Next() {655 var hash, raw string656 if err := rows.Scan(&hash, &raw); err != nil {657 rows.Close()658 return err659 }660 var signed event.Event661 if json.Unmarshal([]byte(raw), &signed) != nil {662 continue663 }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] = sources671 }672 if err := errors.Join(rows.Err(), rows.Close()); err != nil {673 return err674 }675 for hash, sources := range tagged {676 current, err := t.artifactSources(ctx, name, hash)677 if err != nil {678 return err679 }680 sort.Strings(sources)681 sort.Strings(current)682 if strings.Join(sources, "\n") == strings.Join(current, "\n") {683 continue684 }685 if err := t.signArtifact(ctx, view, hash, time.Now().Unix()); err != nil {686 return err687 }688 }689 return nil690}691692func (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, err696 }697 defer rows.Close()698 var out []string699 for rows.Next() {700 var source string701 if err := rows.Scan(&source); err != nil {702 return nil, err703 }704 out = append(out, source)705 }706 return out, rows.Err()707}708709func (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, err713 }714 defer rows.Close()715 var ids []string716 for rows.Next() {717 var id string718 if err := rows.Scan(&id); err != nil {719 return nil, err720 }721 if id != "" {722 ids = append(ids, id)723 }724 }725 return ids, rows.Err()726}727728// 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 err733 }734 }735 return nil736}737738// cascadeViewDeletion runs after a kind 5 deletion is stored: artifacts739// 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 return743 }744 rows, err := t.customViewRows(ctx)745 if err != nil || len(rows) == 0 {746 return747 }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}754755// artifactRecord is one stored artifact as the route serves it.756type artifactRecord struct {757 Type string758 Audience string759 Content string760 EventID string761}762763func (t *customViewService) customViewArtifact(ctx context.Context, name, hash string) (artifactRecord, error) {764 var record artifactRecord765 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, err770}771772// 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), nil778}