model

package
v0.3.16 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Jul 3, 2026 License: MIT Imports: 5 Imported by: 0

Documentation

Overview

Package model is the shared contract for TinyRaven: the data types parsed from .datasource/.pipe files plus the interfaces that decouple the subsystems (gatherer, pipe executor, clickhouse, auth, api) from each other. Packages depend on these interfaces, never on each other's concrete types, so they compile and develop independently.

Index

Constants

This section is empty.

Variables

View Source
var ErrUnknownDatasource = errors.New("unknown datasource")

ErrUnknownDatasource is returned by Ingester when events target a datasource that isn't registered. The API maps it to 404 (ADR 0008 — no schema-on-write).

Functions

This section is empty.

Types

type CHInserter

type CHInserter interface {
	Insert(ctx context.Context, ds *Datasource, rows []map[string]any) error
}

CHInserter does batched native-protocol inserts (clickhouse-go/v2, TCP 9000; ADR 0013). The caller passes the datasource (for column order/types) and rows already parsed + validated against its schema, so the implementation can build a typed native batch (PrepareBatch).

type CHQuerier

type CHQuerier interface {
	Query(ctx context.Context, sql string, params, settings map[string]string) ([]byte, error)
}

CHQuerier runs read-only queries over the ClickHouse HTTP interface (8123; ADR 0013). params are CH query parameters (param_<name> values); settings are extra SETTINGS (e.g. use_query_cache). Returns the response body verbatim (caller asks for FORMAT JSON).

type CHWriter

type CHWriter interface {
	InsertSelect(ctx context.Context, sql string, params map[string]string) error
}

CHWriter runs a write statement (INSERT INTO ... SELECT) over the read-write ClickHouse identity, binding CH params (param_<name> values). Copy pipes use it to land a query's rows in a target datasource; the read-only Query path (readonly=2; ADR 0011) would refuse the INSERT, so this is a separate method.

type Column

type Column struct {
	Name string
	Type string // ClickHouse type verbatim, e.g. "String", "DateTime", "JSON"
}

Column is one SCHEMA column in a .datasource file.

type Copy

type Copy struct {
	Name             string
	TargetDatasource string  // INSERT target; validated to exist at deploy time
	Schedule         string  // cron expression from COPY_SCHEDULE; "" = on-demand only
	SQL              string  // the SELECT whose rows are inserted
	Params           []Param // {{Type(name)}} placeholders in the copy SQL, in order
}

Copy is a TYPE copy pipe (Tinybird copy pipes): it runs its SQL and INSERTs the result into TargetDatasource. Mirrors Materialization, plus an optional cron Schedule (COPY_SCHEDULE), auto-executed by internal/scheduler (standard 5-field cron; gap #8). The copy node may be parameterized, so it carries its own Params (extracted like an Endpoint's) — but a scheduled run supplies no request params, so a required (no-default) param will 400 on trigger; a pipe meant to run on a schedule should give every param a default.

type CopyRunner

type CopyRunner interface {
	RunCopy(ctx context.Context, name string, params url.Values) (body []byte, status int, err error)
}

CopyRunner triggers a TYPE copy pipe on demand: it composes the copy SQL (same param binding/control flow as a query), runs INSERT INTO <target> SELECT ... over the write path, and returns a Tinybird-shaped copy-job JSON body. status is the HTTP status to send. Kept separate from PipeRunner so query-only callers (and their fakes) need not implement it.

type Datasource

type Datasource struct {
	Name       string            // file basename without extension
	Schema     []Column          // ordered, as written
	Engine     string            // "MergeTree" if ENGINE omitted
	EngineOpts map[string]string // ENGINE_* keys (full key incl. prefix) -> value, verbatim
	Raw        string            // original file text
}

Datasource is a parsed .datasource file. ENGINE defaults to "MergeTree ORDER BY tuple()" when omitted (ADR 0008); all ENGINE_* options are forwarded to ClickHouse verbatim.

func (Datasource) QuarantineTable

func (d Datasource) QuarantineTable() string

QuarantineTable is the CH table invalid rows land in (ADR 0018).

type DatasourceRegistry

type DatasourceRegistry interface {
	Get(ctx context.Context, name string) (*Datasource, bool, error)
	Put(ctx context.Context, ds *Datasource) error
	List(ctx context.Context) ([]*Datasource, error)
}

DatasourceRegistry stores parsed datasource schemas (Redis-backed; ADR 0001).

type Endpoint

type Endpoint struct {
	Name      string
	SQL       string
	Params    []Param // extracted from SQL, in order of appearance
	RateLimit int     // requests/sec, 0 = unset
	CacheTTL  int     // seconds; 0 = caching off (ADR 0009)
}

Endpoint is an ENDPOINT block (TYPE query) — a published API route.

type Ingester

type Ingester interface {
	Ingest(ctx context.Context, datasource string, rows []json.RawMessage) (successful, quarantined int, err error)
}

Ingester accepts raw event rows for a datasource, validates + quarantines per row, and buffers the valid ones (ADRs 0004, 0018). It never rejects the batch for bad rows; counts are returned for the 202 body.

type JobRecord added in v0.3.16

type JobRecord struct {
	ID        string
	Kind      string // "copy" today; other job kinds (gap #6 import) would add values
	Status    string // "done" | "error" until an async path exists
	PipeName  string
	Target    string
	CreatedAt time.Time
}

JobRecord is a persisted async-job record (Tinybird job API parity, gap #8). TinyRaven only produces "copy" jobs today, always terminal ("done") since RunCopy executes synchronously; the record exists so job_url / GET /v0/jobs are pollable instead of dead links. Target is the copy's destination datasource, carried so GET authorization can reuse the same APPEND:<target> scope check RunCopy already requires to create the job.

type JobStore added in v0.3.16

type JobStore interface {
	Put(ctx context.Context, rec JobRecord) error
	Get(ctx context.Context, id string) (JobRecord, bool, error)
	// List returns up to limit most-recent records, newest first.
	List(ctx context.Context, limit int) ([]JobRecord, error)
}

JobStore persists JobRecords for GET /v0/jobs and GET /v0/jobs/{id}. Redis-backed with a TTL (jobs have no git-tracked source of truth, unlike DatasourceRegistry/PipeRegistry, so they don't belong in the AOF-persisted metadata registry — ADR 0001 covers config, not ephemeral run history).

type Materialization

type Materialization struct {
	Name        string
	TargetTable string
	SQL         string
}

Materialization is a MATERIALIZATION block — wired in Phase 3, parsed now.

type Node

type Node struct {
	Name string
	SQL  string
}

Node is a NODE block in a .pipe file — a named, reusable SQL fragment.

type OutputFormat

type OutputFormat string

OutputFormat is the response format for a published pipe endpoint. It maps to a ClickHouse FORMAT and an HTTP content type (ADR 0003). Tinybird exposes the format as the URL extension on /v0/pipes/{name}.{format}; the zero value FormatJSON is the default ({meta,data,rows,statistics} envelope).

const (
	FormatJSON    OutputFormat = "json"    // CH FORMAT JSON   -> application/json
	FormatCSV     OutputFormat = "csv"     // CH FORMAT CSVWithNames -> text/csv (header row)
	FormatNDJSON  OutputFormat = "ndjson"  // CH FORMAT JSONEachRow -> application/x-ndjson
	FormatParquet OutputFormat = "parquet" // CH FORMAT Parquet -> application/octet-stream (binary)
)

type Param

type Param struct {
	Name       string
	Type       ParamType
	Default    string
	HasDefault bool
}

Param is a {{Type(name, default)}} placeholder extracted from an ENDPOINT SQL body (ADR 0003). MVP supports value params only.

type ParamType

type ParamType string

ParamType is a templated query parameter type, e.g. String, DateTime, Int64.

type Pinger

type Pinger interface {
	Ping(ctx context.Context) error
}

Pinger is a generic readiness probe (Redis, ClickHouse) — Ping returns nil when the dependency is reachable (ADR 0024).

type Pipe

type Pipe struct {
	Name     string
	Nodes    []Node
	Endpoint *Endpoint
	Material *Materialization
	Copy     *Copy
	Raw      string
}

Pipe is a parsed .pipe file. A pipe has zero or more NODEs and at most one terminal block: an ENDPOINT, a MATERIALIZATION, or a COPY (TYPE copy).

type PipeRegistry

type PipeRegistry interface {
	Get(name string) (*Pipe, bool)
	Put(p *Pipe)
	List() []*Pipe
}

PipeRegistry stores parsed pipes. In-memory with atomic swap on hot reload (ADR 0020).

type PipeRunner

type PipeRunner interface {
	Run(ctx context.Context, name string, params url.Values) (body []byte, status int, err error)
	RunFormat(ctx context.Context, name string, params url.Values, format OutputFormat) (body []byte, status int, err error)
}

PipeRunner executes a published pipe endpoint and returns the response body. status is the HTTP status to send; err is for internal failures. Run returns the default JSON envelope; RunFormat selects the output format (ADR 0003) — only the trailing ClickHouse FORMAT differs, all param binding/validation is shared. Run is retained as the JSON shorthand (RunFormat(..., FormatJSON)).

type QueryStat

type QueryStat struct {
	Pipe       string
	DurationMS float64
	ReadRows   int64
	ReadBytes  int64
	StatusCode int
	Error      string // empty on success
}

QueryStat is one pipe execution's observability record (ADR 0014). Fed through the Gatherer into the tinybird.pipe_stats table, async + best-effort.

type StatsRecorder

type StatsRecorder interface {
	Record(stat QueryStat)
}

StatsRecorder receives per-query stats. Implementations MUST be non-blocking (drop on overflow) so observability never slows the query path (ADR 0014).

type Token

type Token struct {
	Name   string
	Value  string
	Scopes []string
}

Token is an auth token (ADR 0005). Value is the secret bearer string; Name is the git-tracked identifier; Scopes gate access (e.g. "ADMIN", "READ:<pipe>").

func (Token) HasScope

func (t Token) HasScope(s string) bool

HasScope reports whether the token carries scope s or ADMIN.

type TokenManager added in v0.3.16

type TokenManager interface {
	// List returns every stored token. Callers MUST NOT expose Token.Value.
	List(ctx context.Context) ([]*Token, error)
	// Create mints a new opaque bearer value, persists the token, and returns it
	// (Value populated) exactly once — the only time the secret is revealed.
	Create(ctx context.Context, name string, scopes []string) (*Token, error)
	// DeleteByName revokes the token with the given name; ok=false if none matched.
	DeleteByName(ctx context.Context, name string) (ok bool, err error)
}

TokenManager is the ADMIN-only management surface for the /v0/tokens API (parity gap #7). It is deliberately separate from TokenStore: Validate is on the auth hot path (every request), while these are cold, admin-gated operations. Keeping them apart means fakes and alternate stores need not grow management methods to satisfy the auth path.

type TokenStore

type TokenStore interface {
	// Validate returns the token for a bearer value, or ok=false if unknown.
	Validate(ctx context.Context, value string) (*Token, bool, error)
	Put(ctx context.Context, t *Token) error
}

TokenStore validates and manages bearer tokens (Redis-backed; ADR 0005).

type Truncater added in v0.3.16

type Truncater interface {
	Truncate(ctx context.Context, table string) error
}

Truncater empties a datasource's backing table (TRUNCATE TABLE). The batch-import replace mode (POST /v0/datasources?mode=replace) clears the target before loading. Runs over the read-write identity, like the other DDL; kept separate from the insert path so append-only importers (and their fakes) need not implement it, and a nil Truncater simply disables replace mode.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL