1package daemon23// Custom views hand fenced code blocks to an external transform and keep4// what comes back as relay-signed artifacts attached to the source event.5// This file holds the definition, its storage and the owner's management6// methods; custom_views_transform.go holds the block extraction, the POST7// and the artifacts, and custom_views_http.go serves them.89import (10 "context"11 "crypto/sha256"12 "database/sql"13 "encoding/hex"14 "encoding/json"15 "errors"16 "fmt"17 "net/http"18 "sort"19 "strconv"20 "strings"21 "sync"22 "time"2324 "github.com/FelineStateMachine/tinyrelay/internal/community"25 "github.com/FelineStateMachine/tinyrelay/internal/event"26 "github.com/FelineStateMachine/tinyrelay/internal/gates"27 "github.com/FelineStateMachine/tinyrelay/internal/records"28 "github.com/FelineStateMachine/tinyrelay/internal/storage"29 "github.com/FelineStateMachine/tinyrelay/internal/telemetry"30 "github.com/FelineStateMachine/tinyrelay/internal/views"31 gitrelay "github.com/FelineStateMachine/tinyrelay/tinygit"32)3334// customViewService owns custom view definitions, transform execution and35// artifact storage. It borrows tenant dependencies and serializes transforms36// per view. Git is bound after repository initialization and before workers37// start; Records supplies the signing identity for generated artifacts.38type customViewService struct {39 store *storage.Store40 community *community.Service41 records *records.Service42 git *gitrelay.GitRelay43 telemetry *telemetry.Telemetry44 tenantName string45 publicURL string46 loopback func() bool47 resolveActor func(*http.Request) (string, error)48 browseRead func(context.Context, string) error49 gate *gates.Gate50 viewClient *http.Client51 viewMu sync.Mutex52 viewIndex map[int][]customView53 viewLocks map[string]*sync.Mutex54}5556type customViewServiceConfig struct {57 Store *storage.Store58 Community *community.Service59 Records *records.Service60 Git *gitrelay.GitRelay61 Telemetry *telemetry.Telemetry62 TenantName string63 PublicURL string64 Loopback func() bool65 ResolveActor func(*http.Request) (string, error)66 BrowseRead func(context.Context, string) error67 Gate *gates.Gate68}6970func newCustomViewService(cfg customViewServiceConfig) *customViewService {71 return &customViewService{72 store: cfg.Store,73 community: cfg.Community,74 records: cfg.Records,75 git: cfg.Git,76 telemetry: cfg.Telemetry,77 tenantName: cfg.TenantName,78 publicURL: cfg.PublicURL,79 loopback: cfg.Loopback,80 resolveActor: cfg.ResolveActor,81 browseRead: cfg.BrowseRead,82 gate: cfg.Gate,83 }84}8586const (87 viewTransform = "view-transform"88 viewKindMax = 1689 viewLanguageMax = 3290 viewLanguageLength = 3291 viewDefaultMaxBytes = 1 << 2092 viewMaxBytesCap = 4 << 2093 viewTimeout = 30 * time.Second94 viewMaxAttempts = 395 viewPauseFailures = 2096 viewBackfillLimit = 50097 viewHourlyPeriod = int64(3600)98 viewResponseMaxBytes = int64(viewMaxBytesCap*4 + 1<<20)99)100101// viewBackoff is the wait before the second and third attempt.102var viewBackoff = []time.Duration{time.Minute, 5 * time.Minute}103104const customViewSchema = `CREATE TABLE IF NOT EXISTS custom_views(name TEXT PRIMARY KEY, kinds TEXT NOT NULL, transform TEXT NOT NULL, trigger TEXT NOT NULL, audience TEXT NOT NULL, languages TEXT NOT NULL, max_bytes INTEGER NOT NULL, secret TEXT NOT NULL, generation TEXT NOT NULL DEFAULT '', enabled INTEGER NOT NULL DEFAULT 1, created_at INTEGER NOT NULL, last_run_at INTEGER NOT NULL DEFAULT 0, last_status TEXT NOT NULL DEFAULT '', failures INTEGER NOT NULL DEFAULT 0);105CREATE TABLE IF NOT EXISTS custom_view_artifacts(view TEXT NOT NULL, hash TEXT NOT NULL, event_id TEXT NOT NULL, type TEXT NOT NULL, engine TEXT NOT NULL DEFAULT '', audience TEXT NOT NULL, content TEXT NOT NULL, raw TEXT NOT NULL, created_at INTEGER NOT NULL, PRIMARY KEY(view,hash));106CREATE TABLE IF NOT EXISTS custom_view_sources(view TEXT NOT NULL, hash TEXT NOT NULL, source TEXT NOT NULL, block INTEGER NOT NULL, expires INTEGER NOT NULL DEFAULT 0, PRIMARY KEY(view,hash,source));107CREATE INDEX IF NOT EXISTS custom_view_sources_source ON custom_view_sources(view,source);`108109// customView is one stored definition. The secret never leaves the tenant110// except once, in the addcustomview answer.111type customView struct {112 Name string `json:"name"`113 Kinds []int `json:"kinds"`114 Transform string `json:"transform"`115 Trigger string `json:"trigger"`116 Audience string `json:"audience"`117 Languages []string `json:"languages"`118 MaxBytes int `json:"maxBytes"`119 Enabled bool `json:"enabled"`120 CreatedAt int64 `json:"created"`121 LastRunAt int64 `json:"lastRun"`122 LastStatus string `json:"lastStatus"`123 Failures int `json:"failures"`124 secret string125 generation string126}127128type customViewTarget struct {129 Name string130 Fingerprint string131}132133func customViewFingerprint(view customView) string {134 canonical := struct {135 Name string136 Kinds []int137 Transform string138 Trigger string139 Audience string140 Languages []string141 MaxBytes int142 Secret string143 CreatedAt int64144 Generation string145 }{view.Name, view.Kinds, view.Transform, view.Trigger, view.Audience, view.Languages, view.MaxBytes, view.secret, view.CreatedAt, view.generation}146 raw, _ := json.Marshal(canonical)147 sum := sha256.Sum256(raw)148 return hex.EncodeToString(sum[:])149}150151func (v customView) summary() map[string]any {152 state := "active"153 if !v.Enabled {154 state = "paused"155 }156 return map[string]any{"name": v.Name, "kinds": v.Kinds, "transform": v.Transform, "host": callbackHost(v.Transform), "trigger": v.Trigger, "audience": v.Audience, "languages": v.Languages, "maxBytes": v.MaxBytes, "enabled": v.Enabled, "state": state, "created": v.CreatedAt, "lastRun": v.LastRunAt, "lastStatus": v.LastStatus, "failures": v.Failures}157}158159func (v customView) watches(kind int) bool {160 for _, k := range v.Kinds {161 if k == kind {162 return true163 }164 }165 return false166}167168func (t *customViewService) initCustomViews(ctx context.Context) error {169 if _, err := t.store.DB().ExecContext(ctx, customViewSchema); err != nil {170 return err171 }172 _, err := t.store.DB().ExecContext(ctx, `ALTER TABLE custom_views ADD COLUMN generation TEXT NOT NULL DEFAULT ''`)173 if err != nil && !strings.Contains(strings.ToLower(err.Error()), "duplicate column") {174 return err175 }176 rows, err := t.store.DB().QueryContext(ctx, `SELECT name FROM custom_views WHERE generation=''`)177 if err != nil {178 return err179 }180 defer rows.Close()181 var names []string182 for rows.Next() {183 var name string184 if err := rows.Scan(&name); err != nil {185 return err186 }187 names = append(names, name)188 }189 if err := rows.Err(); err != nil {190 return err191 }192 for _, name := range names {193 generation, err := randomHex(16)194 if err != nil {195 return err196 }197 if _, err := t.store.DB().ExecContext(ctx, `UPDATE custom_views SET generation=? WHERE name=? AND generation=''`, generation, name); err != nil {198 return err199 }200 }201 return nil202}203204func customViewMethod(method string) bool {205 return containsString([]string{"listcustomviews", "addcustomview", "removecustomview", "runcustomview", "pausecustomview", "resumecustomview"}, method)206}207208// customViewExecute serves the owner's custom view methods under the view209// operation. Every change is recorded in the audit log.210func (t *customViewService) customViewExecute(ctx context.Context, actor, method string, params []json.RawMessage) (result any, err error) {211 ctx, finish := t.telemetry.Start(ctx, "view")212 outcome := "error"213 defer func() {214 if err != nil {215 outcome = callbackOutcome(err)216 }217 finish(outcome)218 }()219 role, err := t.community.Role(ctx, actor)220 if err != nil {221 return nil, err222 }223 if role != "owner" {224 return nil, fmt.Errorf("restricted: only the owner may %s", method)225 }226 switch method {227 case "listcustomviews":228 rows, err := t.customViewRows(ctx)229 if err != nil {230 return nil, err231 }232 out := make([]map[string]any, 0, len(rows))233 for _, row := range rows {234 out = append(out, row.summary())235 }236 outcome = "ok"237 return out, nil238 case "addcustomview":239 result, err = t.addCustomView(ctx, actor, params)240 default:241 result, err = t.changeCustomView(ctx, actor, method, params)242 }243 if err == nil {244 outcome = "ok"245 }246 return result, err247}248249// customViewOptions is the addcustomview parameter. Kinds and languages may250// arrive as lists or as comma-separated strings from a plain form.251type customViewOptions struct {252 Name string `json:"name"`253 Kinds json.RawMessage `json:"kinds"`254 Transform string `json:"transform"`255 Trigger string `json:"trigger"`256 Audience string `json:"audience"`257 Languages json.RawMessage `json:"languages"`258 MaxBytes json.RawMessage `json:"max_bytes"`259 Secret string `json:"secret"`260}261262func (t *customViewService) addCustomView(ctx context.Context, actor string, params []json.RawMessage) (any, error) {263 if len(params) != 1 {264 return nil, errors.New("invalid: addcustomview expects one object with name, kinds, transform, trigger, audience, languages and optional max_bytes and secret")265 }266 var options customViewOptions267 if err := json.Unmarshal(params[0], &options); err != nil {268 return nil, errors.New("invalid: addcustomview expects one object with name, kinds, transform, trigger, audience, languages and optional max_bytes and secret")269 }270 view, err := t.parseCustomView(options)271 if err != nil {272 return nil, err273 }274 generated := false275 secret := strings.TrimSpace(options.Secret)276 if secret == "" {277 secret, err = randomHex(32)278 if err != nil {279 return nil, err280 }281 generated = true282 } else if len(secret) < callbackSecretMin || len(secret) > callbackSecretMax || strings.Trim(secret, "!\"#$%&'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijklmnopqrstuvwxyz{|}~") != "" {283 return nil, fmt.Errorf("invalid: secret must be %d to %d printable ASCII characters", callbackSecretMin, callbackSecretMax)284 }285 view.secret = secret286 view.generation, err = randomHex(16)287 if err != nil {288 return nil, err289 }290 view.Enabled = true291 view.CreatedAt = time.Now().Unix()292 kinds, _ := json.Marshal(view.Kinds)293 languages, _ := json.Marshal(view.Languages)294 _, err = t.store.DB().ExecContext(ctx, `INSERT INTO custom_views(name,kinds,transform,trigger,audience,languages,max_bytes,secret,generation,enabled,created_at) VALUES(?,?,?,?,?,?,?,?,?,1,?)`, view.Name, string(kinds), view.Transform, view.Trigger, view.Audience, string(languages), view.MaxBytes, secret, view.generation, view.CreatedAt)295 if err != nil {296 if strings.Contains(err.Error(), "UNIQUE") {297 return nil, errors.New("invalid: a view with that name exists")298 }299 return nil, err300 }301 t.invalidateCustomViews()302 if err := t.community.Record(ctx, actor, "addcustomview", view.Name, ""); err != nil {303 return nil, err304 }305 // Counts only: the transform host, path and secret stay out of the log.306 t.telemetry.Logger().Info("custom view added", "tenant", t.tenantName, "kinds", len(view.Kinds), "languages", len(view.Languages), "generated", generated)307 out := view.summary()308 out["secret"] = secret309 return out, nil310}311312// parseCustomView validates a definition: the name, the kinds it watches,313// the https transform, the trigger, the audience, the languages and the314// artifact size limit.315func (t *customViewService) parseCustomView(options customViewOptions) (customView, error) {316 view := customView{Name: strings.TrimSpace(options.Name), Transform: strings.TrimSpace(options.Transform), Trigger: strings.TrimSpace(options.Trigger), Audience: strings.TrimSpace(options.Audience), MaxBytes: viewDefaultMaxBytes}317 if !views.NamePattern.MatchString(view.Name) {318 return customView{}, errors.New("invalid: name must be 1 to 32 lowercase letters, digits or hyphens")319 }320 kinds, err := intList(options.Kinds)321 if err != nil || len(kinds) == 0 || len(kinds) > viewKindMax {322 return customView{}, fmt.Errorf("invalid: kinds must list 1 to %d event kinds", viewKindMax)323 }324 for _, kind := range kinds {325 if kind < 0 || kind > 65535 {326 return customView{}, errors.New("invalid: kinds must be between 0 and 65535")327 }328 }329 sort.Ints(kinds)330 view.Kinds = dedupeInts(kinds)331 if err := validateWebhookURL(view.Transform, t.publicURL); err != nil {332 return customView{}, fmt.Errorf("invalid: transform %s", strings.TrimPrefix(err.Error(), "invalid: "))333 }334 if view.Trigger == "" {335 view.Trigger = "write"336 }337 if view.Trigger != "write" && view.Trigger != "hourly" {338 return customView{}, errors.New("invalid: trigger must be write or hourly")339 }340 if view.Audience == "" {341 view.Audience = "public"342 }343 if view.Audience != "public" && view.Audience != "members" {344 return customView{}, errors.New("invalid: audience must be public or members")345 }346 languages, err := stringList(options.Languages)347 if err != nil || len(languages) == 0 || len(languages) > viewLanguageMax {348 return customView{}, fmt.Errorf("invalid: languages must list 1 to %d fenced block languages", viewLanguageMax)349 }350 seen := map[string]bool{}351 view.Languages = nil352 for _, lang := range languages {353 lang = strings.ToLower(strings.TrimSpace(lang))354 if lang == "" || len(lang) > viewLanguageLength || strings.Trim(lang, "abcdefghijklmnopqrstuvwxyz0123456789+-_.#") != "" {355 return customView{}, errors.New("invalid: languages must be lowercase words of letters, digits, +, -, _, . or #")356 }357 if !seen[lang] {358 seen[lang] = true359 view.Languages = append(view.Languages, lang)360 }361 }362 if len(options.MaxBytes) > 0 && string(options.MaxBytes) != "null" && string(options.MaxBytes) != `""` {363 var size int364 var text string365 if json.Unmarshal(options.MaxBytes, &size) != nil {366 if json.Unmarshal(options.MaxBytes, &text) != nil {367 return customView{}, errors.New("invalid: max_bytes must be a number")368 }369 if size, err = strconv.Atoi(strings.TrimSpace(text)); err != nil {370 return customView{}, errors.New("invalid: max_bytes must be a number")371 }372 }373 if size < 1 || size > viewMaxBytesCap {374 return customView{}, fmt.Errorf("invalid: max_bytes must be between 1 and %d", viewMaxBytesCap)375 }376 view.MaxBytes = size377 }378 return view, nil379}380381// intList reads a JSON list of integers, a list of numeric strings or a382// comma-separated string.383func intList(raw json.RawMessage) ([]int, error) {384 if len(raw) == 0 {385 return nil, nil386 }387 var out []int388 if json.Unmarshal(raw, &out) == nil {389 return out, nil390 }391 items, err := stringList(raw)392 if err != nil {393 return nil, err394 }395 for _, item := range items {396 value, err := strconv.Atoi(strings.TrimSpace(item))397 if err != nil {398 return nil, err399 }400 out = append(out, value)401 }402 return out, nil403}404405// stringList reads a JSON list of strings or a comma-separated string.406func stringList(raw json.RawMessage) ([]string, error) {407 if len(raw) == 0 {408 return nil, nil409 }410 var out []string411 if json.Unmarshal(raw, &out) == nil {412 return out, nil413 }414 var text string415 if err := json.Unmarshal(raw, &text); err != nil {416 return nil, err417 }418 for _, item := range strings.FieldsFunc(text, func(r rune) bool { return r == ',' || r == ' ' || r == '\n' }) {419 if item = strings.TrimSpace(item); item != "" {420 out = append(out, item)421 }422 }423 return out, nil424}425426func dedupeInts(values []int) []int {427 out := values[:0]428 for i, value := range values {429 if i == 0 || value != values[i-1] {430 out = append(out, value)431 }432 }433 return out434}435436func (t *customViewService) changeCustomView(ctx context.Context, actor, method string, params []json.RawMessage) (any, error) {437 name := customViewNameParam(params)438 if name == "" {439 return nil, errors.New("invalid: view name required")440 }441 view, err := t.customViewByName(ctx, name)442 if err != nil {443 return nil, err444 }445 switch method {446 case "removecustomview":447 if err := t.removeCustomView(ctx, view); err != nil {448 return nil, err449 }450 case "pausecustomview":451 _, err = t.store.DB().ExecContext(ctx, `UPDATE custom_views SET enabled=0, last_status='paused' WHERE name=?`, name)452 view.Enabled, view.LastStatus = false, "paused"453 case "resumecustomview":454 _, err = t.store.DB().ExecContext(ctx, `UPDATE custom_views SET enabled=1, failures=0, last_status='' WHERE name=?`, name)455 view.Enabled, view.Failures, view.LastStatus = true, 0, ""456 case "runcustomview":457 if !view.Enabled {458 return nil, errors.New("invalid: resume the view before running it")459 }460 refresh := customViewRefreshParam(params)461 queued, err := t.queueCustomViewBackfill(ctx, view, 0, strconv.FormatInt(time.Now().UnixNano(), 10), refresh)462 if err != nil {463 return nil, err464 }465 if err := t.community.Record(ctx, actor, method, name, strconv.Itoa(queued)); err != nil {466 return nil, err467 }468 t.telemetry.Logger().Info("custom view run queued", "tenant", t.tenantName, "queued", queued)469 out := view.summary()470 out["queued"] = queued471 return out, nil472 default:473 return nil, fmt.Errorf("unsupported: management method %q", method)474 }475 if err != nil {476 return nil, err477 }478 t.invalidateCustomViews()479 if err := t.community.Record(ctx, actor, method, name, ""); err != nil {480 return nil, err481 }482 if method == "removecustomview" {483 return map[string]any{"name": name, "removed": true}, nil484 }485 return view.summary(), nil486}487488// removeCustomView deletes the definition and every artifact it produced,489// including the stored public records.490func (t *customViewService) removeCustomView(ctx context.Context, view customView) error {491 ids, err := t.customViewArtifactIDs(ctx, `SELECT event_id FROM custom_view_artifacts WHERE view=?`, view.Name)492 if err != nil {493 return err494 }495 if err := t.store.WithTx(ctx, func(tx *sql.Tx) error {496 for _, table := range []string{"custom_view_sources", "custom_view_artifacts", "custom_views"} {497 column := "view"498 if table == "custom_views" {499 column = "name"500 }501 if _, err := tx.ExecContext(ctx, "DELETE FROM "+table+" WHERE "+column+"=?", view.Name); err != nil {502 return err503 }504 }505 return nil506 }); err != nil {507 return err508 }509 return t.deleteStoredArtifacts(ctx, ids)510}511512// customViewNameParam reads the name from a bare string or from {"name": ...}.513func customViewNameParam(params []json.RawMessage) string {514 if len(params) == 0 {515 return ""516 }517 if name := stringParam(params, 0); name != "" {518 return name519 }520 var options struct {521 Name string `json:"name"`522 }523 _ = json.Unmarshal(params[0], &options)524 return strings.TrimSpace(options.Name)525}526527func customViewRefreshParam(params []json.RawMessage) bool {528 if len(params) == 0 {529 return false530 }531 var options struct {532 Refresh bool `json:"refresh"`533 }534 if err := json.Unmarshal(params[0], &options); err != nil {535 return false536 }537 return options.Refresh538}539540func (t *customViewService) customViewByName(ctx context.Context, name string) (customView, error) {541 rows, err := t.customViewRows(ctx, name)542 if err != nil {543 return customView{}, err544 }545 if len(rows) == 0 {546 return customView{}, errors.New("not found: view")547 }548 return rows[0], nil549}550551// customViewRows lists definitions, narrowed to one name.552func (t *customViewService) customViewRows(ctx context.Context, names ...string) ([]customView, error) {553 query := `SELECT name,kinds,transform,trigger,audience,languages,max_bytes,secret,generation,enabled,created_at,last_run_at,last_status,failures FROM custom_views`554 args := []any{}555 if len(names) > 0 {556 query += ` WHERE name=?`557 args = append(args, names[0])558 }559 query += ` ORDER BY created_at,rowid`560 rows, err := t.store.DB().QueryContext(ctx, query, args...)561 if err != nil {562 return nil, fmt.Errorf("list custom views: %w", err)563 }564 defer rows.Close()565 var out []customView566 for rows.Next() {567 var item customView568 var kinds, languages string569 var enabled int570 if err := rows.Scan(&item.Name, &kinds, &item.Transform, &item.Trigger, &item.Audience, &languages, &item.MaxBytes, &item.secret, &item.generation, &enabled, &item.CreatedAt, &item.LastRunAt, &item.LastStatus, &item.Failures); err != nil {571 return nil, err572 }573 item.Enabled = enabled != 0574 if err := json.Unmarshal([]byte(kinds), &item.Kinds); err != nil {575 return nil, fmt.Errorf("custom view kinds: %w", err)576 }577 if err := json.Unmarshal([]byte(languages), &item.Languages); err != nil {578 return nil, fmt.Errorf("custom view languages: %w", err)579 }580 out = append(out, item)581 }582 return out, rows.Err()583}584585// customViewIndex returns the enabled views indexed by kind, built on first586// use and dropped whenever a view changes.587func (t *customViewService) customViewIndex(ctx context.Context) map[int][]customView {588 t.viewMu.Lock()589 defer t.viewMu.Unlock()590 if t.viewIndex != nil {591 return t.viewIndex592 }593 index := map[int][]customView{}594 rows, err := t.customViewRows(ctx)595 if err != nil {596 t.telemetry.Logger().Error("load custom views", "tenant", t.tenantName, "error", err)597 return index598 }599 for _, row := range rows {600 if !row.Enabled {601 continue602 }603 for _, kind := range row.Kinds {604 index[kind] = append(index[kind], row)605 }606 }607 t.viewIndex = index608 return index609}610611func (t *customViewService) invalidateCustomViews() {612 t.viewMu.Lock()613 t.viewIndex = nil614 t.viewMu.Unlock()615}616617// CustomViews is the summary the web UI renders with: the enabled views'618// names and languages, nothing else.619func (t *customViewService) CustomViews() []views.View {620 seen := map[string]bool{}621 var out []views.View622 for _, list := range t.customViewIndex(context.Background()) {623 for _, view := range list {624 if !seen[view.Name] {625 seen[view.Name] = true626 out = append(out, views.View{Name: view.Name, Languages: append([]string(nil), view.Languages...)})627 }628 }629 }630 sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })631 return out632}633634// queueCustomViews runs after an event is stored or released. Each enabled635// write-triggered view that watches the kind and finds a block it renders636// gets one transform intent. It never fails the publish.637func (t *customViewService) queueCustomViews(ctx context.Context, e event.Event) {638 intents := t.PrepareReleased(ctx, e)639 if len(intents) == 0 {640 return641 }642 err := t.store.WithTx(ctx, func(tx *sql.Tx) error {643 return storage.AddIntents(ctx, tx, intents, time.Now().Unix())644 })645 if err != nil {646 t.telemetry.Logger().Error("view transforms not queued", "tenant", t.tenantName, "error", err)647 return648 }649 t.telemetry.Logger().Debug("view transforms queued", "tenant", t.tenantName, "queued", len(intents))650}651652// Prepare returns write-triggered transform intents for an event. The caller653// owns the transaction that persists the returned intents, allowing event and654// follow-up work to share one commit boundary.655func (t *customViewService) Prepare(ctx context.Context, e event.Event) []storage.Intent {656 targets, err := t.CandidateTargets(ctx, e)657 if err != nil {658 t.telemetry.Logger().Error("plan custom view transforms", "tenant", t.tenantName, "error", err)659 return nil660 }661 intents, err := t.PrepareTargets(ctx, e, targets, false)662 if err != nil {663 t.telemetry.Logger().Error("prepare custom view transforms", "tenant", t.tenantName, "error", err)664 return nil665 }666 return intents667}668669// PrepareReleased returns intents after repository promotion has made Git670// objects available. Repository state events are intentionally accepted only671// through this path.672func (t *customViewService) PrepareReleased(ctx context.Context, e event.Event) []storage.Intent {673 targets, err := t.CandidateTargets(ctx, e)674 if err != nil {675 t.telemetry.Logger().Error("plan released custom view transforms", "tenant", t.tenantName, "error", err)676 return nil677 }678 intents, err := t.PrepareTargets(ctx, e, targets, true)679 if err != nil {680 t.telemetry.Logger().Error("prepare released custom view transforms", "tenant", t.tenantName, "error", err)681 return nil682 }683 return intents684}685686// CandidateTargets captures names and registration identities for an event.687// The identity prevents a later registration with the same name from688// receiving work planned for an earlier registration.689func (t *customViewService) CandidateTargets(ctx context.Context, e event.Event) ([]customViewTarget, error) {690 if eventExpired(e, time.Now().Unix()) {691 return nil, nil692 }693 rows, err := t.customViewRows(ctx)694 if err != nil {695 return nil, err696 }697 targets := make([]customViewTarget, 0, len(rows))698 for _, view := range rows {699 if !view.Enabled || view.Trigger != "write" || !view.watches(e.Kind) {700 continue701 }702 if e.Kind != event.KIND_REPO_STATE && len(views.Matching(views.Blocks(e.Content), view.Languages)) == 0 {703 continue704 }705 targets = append(targets, customViewTarget{Name: view.Name, Fingerprint: customViewFingerprint(view)})706 }707 return targets, nil708}709710// PrepareTargets rechecks captured registration identities before creating711// intents. A removed and recreated view with the same name is rejected.712func (t *customViewService) PrepareTargets(ctx context.Context, e event.Event, targets []customViewTarget, released bool) ([]storage.Intent, error) {713 if len(targets) == 0 || eventExpired(e, time.Now().Unix()) || (!released && e.Kind == event.KIND_REPO_STATE) {714 return nil, nil715 }716 if e.Kind != event.KIND_REPO_STATE {717 current, err := t.sourceCurrent(ctx, e)718 if err != nil {719 return nil, err720 }721 if !current {722 return nil, nil723 }724 }725 rows, err := t.customViewRows(ctx)726 if err != nil {727 return nil, err728 }729 candidates := make(map[string]customView, len(rows))730 for _, view := range rows {731 candidates[view.Name] = view732 }733 if e.Kind == event.KIND_REPO_STATE {734 if t.git == nil || t.statePending(ctx, e.ID) {735 return nil, nil736 }737 }738 var intents []storage.Intent739 for _, target := range targets {740 view, ok := candidates[target.Name]741 if !ok || !view.Enabled || view.Trigger != "write" || !view.watches(e.Kind) {742 continue743 }744 if customViewFingerprint(view) != target.Fingerprint {745 continue746 }747 if e.Kind != event.KIND_REPO_STATE && len(views.Matching(views.Blocks(e.Content), view.Languages)) == 0 {748 continue749 }750 intents = append(intents, t.viewIntent(view, e, "", false))751 }752 return intents, nil753}754755func eventExpired(e event.Event, now int64) bool {756 expires := event.Expiration(e)757 return expires > 0 && expires <= now758}759760// viewIntent is the transform intent for one source. The event travels in761// the payload; run makes a backfill or an hourly pass distinct from the762// write-time intent for the same event.763func (t *customViewService) viewIntent(view customView, e event.Event, run string, refresh ...bool) storage.Intent {764 force := len(refresh) > 0 && refresh[0]765 payload, _ := json.Marshal(viewPayload{Event: e, Attempt: 1, Run: run, Refresh: force, Generation: view.generation})766 id := e.ID767 if run != "" {768 id += "@" + run769 }770 return storage.Intent{Kind: viewTransform, EventID: id, Target: view.Name, Payload: string(payload)}771}772773// statePending reports whether a repository state still waits for its774// objects.775func (t *customViewService) statePending(ctx context.Context, id string) bool {776 var one int777 err := t.store.DB().QueryRowContext(ctx, `SELECT 1 FROM pending_events WHERE id=?`, id).Scan(&one)778 return err == nil || t.git.IsPending(id)779}780781// queueCustomViewBackfill queues the newest sources of a view's kinds, at782// most 500, written since the given time. Repository states are queued783// whether or not their README has blocks; the handler reads it.784func (t *customViewService) queueCustomViewBackfill(ctx context.Context, view customView, since int64, run string, refresh ...bool) (int, error) {785 force := len(refresh) > 0 && refresh[0]786 rows, err := t.store.Query(ctx, event.Filter{Kinds: view.Kinds, Tags: map[string][]string{}}, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}, Limit: viewBackfillLimit})787 if err != nil {788 return 0, err789 }790 var intents []storage.Intent791 for _, e := range rows.Events {792 if e.CreatedAt < since {793 continue794 }795 if e.Kind == event.KIND_REPO_STATE {796 if t.git == nil || t.statePending(ctx, e.ID) {797 continue798 }799 } else if len(views.Matching(views.Blocks(e.Content), view.Languages)) == 0 {800 continue801 }802 intents = append(intents, t.viewIntent(view, e, run, force))803 }804 if len(intents) == 0 {805 return 0, nil806 }807 err = t.store.WithTx(ctx, func(tx *sql.Tx) error {808 return storage.AddIntents(ctx, tx, intents, time.Now().Unix())809 })810 return len(intents), err811}812813// tickCustomViews queues the hourly views that are due. It is called from814// the tenant scheduler on its own timer.815func (t *customViewService) tickCustomViews(ctx context.Context, now int64) error {816 rows, err := t.customViewRows(ctx)817 if err != nil {818 return err819 }820 for _, view := range rows {821 if !view.Enabled || view.Trigger != "hourly" || now-view.LastRunAt < viewHourlyPeriod {822 continue823 }824 since := view.LastRunAt - 300825 if view.LastRunAt == 0 {826 since = 0827 }828 queued, err := t.queueCustomViewBackfill(ctx, view, since, strconv.FormatInt(now/viewHourlyPeriod, 10), false)829 if err != nil {830 return err831 }832 if _, err := t.store.DB().ExecContext(ctx, `UPDATE custom_views SET last_run_at=? WHERE name=?`, now, view.Name); err != nil {833 return err834 }835 if queued > 0 {836 t.telemetry.Logger().Info("custom view hourly run queued", "tenant", t.tenantName, "queued", queued)837 }838 }839 return nil840}