model

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). The copy node may be parameterized, so it carries its own Params (extracted like an Endpoint's).

ponytail: Schedule is parsed and surfaced (introspection, tr deploy) but NOT auto-executed — there is no in-process cron scheduler yet. On-demand triggering via POST /v0/pipes/{name}/copy is the implemented path; scheduled runs are a deferred follow-up (would need a scheduler + the /v0/jobs surface, gap #8).

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 (d Datasource) QuarantineTable() string

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 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 (t Token) HasScope(s string) bool

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).