orm

package module
v1.0.2 Latest Latest
Warning

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

 Go to latest Published: Nov 7, 2026 License: MIT Imports: 41 Imported by: 0

README

Droplet ORM

Code-generation-based Go ORM for MySQL, Redis 8.2+ / Valkey 9.0+, ClickHouse and NATS JetStream (CDC and tasks).

Module: github.com/DropletLive/orm, package orm. Forked from github.com/latolukasz/fluxaorm/v2 at v2.4.4; see "Migrating from fluxaorm" in CLAUDE.md.

Documentation

Index

Constants

View Source 
const (
	// CDCOutboxPending is a row whose event is not yet known to be on the wire.
	CDCOutboxPending = "pending"
	// CDCOutboxDispatched is a row whose event was published successfully.
	CDCOutboxDispatched = "dispatched"
	// CDCOutboxStored is a row from an entity tagged `outbox` without
	// `dirty=`. There is no stream to publish to, so it is born terminal.
	CDCOutboxStored = "stored"
)
View Source 
const (
	JobRunPending      = "pending"
	JobRunRunning      = "running"
	JobRunSucceeded    = "succeeded"
	JobRunFailed       = "failed"
	JobRunDeduplicated = "deduplicated"
)
View Source 
const (
	// EntityStreamName carries every entity change and every replay. One stream
	// rather than one per consumer: the subject already identifies the entity, so
	// per-consumer streams would only re-add the duplication.
	EntityStreamName = "ORM_ENTITY"
	// TaskStreamName carries dispatched tasks. Separate from entity changes
	// because the two need different retention, and because a task backlog must
	// not evict a change nobody has consumed yet.
	TaskStreamName = "ORM_TASK"

	// DefaultQueue carries every task that does not declare a Queue().
	DefaultQueue Queue = "default"
)
View Source 
const DefaultPoolCode = "default"
View Source 
const EntityCacheTTL = time.Hour

EntityCacheTTL bounds how long a cached entity row may disagree with MySQL if an invalidation is ever lost. Correctness comes from the invalidation; this is only the backstop.

View Source 
const HeaderDirtyOp = "Dirty-Op"

HeaderDirtyOp carries the numeric DirtyOp value as a decimal string. It is the only header on an entity message: the entity's identity is the subject, so duplicating it in a header would just be a second thing to keep in sync.

View Source 
const (
	// HeaderJobRunID carries the job_runs row id. It is the one header a task
	// message needs: the task's own identity is the subject, but a row id is not
	// derivable from anything on the wire.
	HeaderJobRunID = "Orm-Job-Run-ID"
)
View Source 
const MaxCachedAllRows = 10000

MaxCachedAllRows caps what `orm:"cached"` will put in one Redis list. Past it GetAll still answers correctly, straight from MySQL - the tag is meant for small reference tables, and a runaway one degrades instead of parking a huge list in Redis for an hour.

View Source 
const MetricsMetaKey = "MetrictMetaKey"
View Source 
const RowCacheRewriteScript = `
redis.call('DEL', KEYS[1])
redis.call('RPUSH', KEYS[1], unpack(ARGV, 2))
redis.call('EXPIRE', KEYS[1], ARGV[1])
return 1
`

RowCacheRewriteScript replaces a cached row's value list in one step. RedisPipeLine is a batch, not a transaction, so two interleaved DEL+RPUSH+EXPIRE runs can leave several lists under one key.

Variables

View Source 
var ErrDispatchInTransaction = errors.New(
	"task dispatch inside a transaction: the publish cannot roll back with it, so dispatch after commit")

ErrDispatchInTransaction is returned when a dispatch is attempted inside an open transaction. The publish is not transactional, so a transaction that later rolls back would leave a task running against data that never existed. Dispatch after the commit instead.

View Source 
var ErrEntityNeedsRegeneration = errors.New("generated entity needs regeneration with orm.Generate to support transactional write snapshots")

ErrEntityNeedsRegeneration rejects generated code without transactional state snapshots. Continuing with that code would silently lose repeated writes.

View Source 
var ErrEntityNotPersisted = errors.New("entity was never persisted")

ErrEntityNotPersisted is returned when Delete is called on an entity that was never written to the database.

View Source 
var ErrEntityReadOnly = errors.New("entity is a read-only write snapshot")

ErrEntityReadOnly protects the frozen entity views passed to after callbacks. Obtain the live entity from its provider to make another write.

View Source 
var ErrEntityUnsavedChanges = errors.New("entity has unsaved changes")

ErrEntityUnsavedChanges is returned by Reload for an entity holding unsaved changes. Reload discards them, and doing that to a handle an outer caller is still mutating is a silent lost update - save or drop the handle first.

View Source 
var ErrEntityVanished = errors.New("entity row no longer exists")

ErrEntityVanished is returned by Reload when the row is gone from the database, so there is nothing left to refresh the handle from.

View Source 
var ErrJobRunEntityNotRegistered = errors.New(
	"orm.JobRunEntity is not registered; add registry.RegisterEntity(orm.JobRunEntity{})")

ErrJobRunEntityNotRegistered is returned by every job-run function when JobRunEntity is absent from the registry. Unlike the outbox functions, which no-op in that case, these fail loudly: a job runner with no run rows has silently lost its entire status history, which is the one thing the table exists to provide.

View Source 
var ErrOutboxDispatchMark = errors.New("cdc outbox dispatch mark failed")

ErrOutboxDispatchMark wraps a failure to flip staged outbox rows to dispatched. The rows are safe: they stay pending, the relay republishes them, and JetStream's dedup window absorbs the duplicate. Distinguishable from a publish failure so a caller can tell "not delivered" from "delivered twice".

View Source 
var ErrPayloadUndecodable = errors.New("task payload cannot be decoded")

ErrPayloadUndecodable marks a payload that will never decode. A task consumer terminates such a message on its first attempt instead of retrying it to the cap, because the bytes on the stream are not going to improve.

View Source 
var ErrTxRollbackOnly = errors.New("transaction is rollback-only")

ErrTxRollbackOnly is returned by the outermost Transaction when a nested Transaction failed and the caller swallowed its error.

Functions

func CDCOutboxBacklog 

func CDCOutboxBacklog(ctx Context) (depth int, oldest time.Duration, err error)

CDCOutboxBacklog reports how many events are undelivered and how old the oldest is. Both are alertable: a stalled relay is otherwise completely silent, because the rows are safe and nothing errors.

func ConsumerPending 

func ConsumerPending(ctx Context, consumer ConsumerName) (uint64, error)

ConsumerPending reports how many messages a consumer still has to deliver.

This is the backlog signal - "is work piling up on this consumer" - and it lives here because the durable name is orm's to decide, so an application asking JetStream directly would have to hardcode the convention.

func DispatchTask 

func DispatchTask[T any](ctx Context, task *T, opts ...DispatchOption) (uint64, error)

DispatchTask publishes one task and returns the ID of its run row.

The row is written before the publish so its ID can ride on the message; a publish the stream deduplicates away closes that row out as deduplicated rather than leaving it pending for a message that will never arrive.

func Generate 

func Generate(engine Engine, outputDirectory string) error

func JobRunBacklog 

func JobRunBacklog(ctx Context) (depth int, oldest time.Duration, err error)

JobRunBacklog reports how many runs are still pending and how old the oldest is. This is the alarm for the one hole in JetStream-only delivery: a row pending for longer than its task's whole backoff ladder means its message was never delivered, and nothing else in the system will say so.

It is not a queue-depth gauge - JetStream's NumPending is that. A healthy busy queue keeps this near zero, because a running job leaves pending immediately.

func JsonMarshalToString 

func JsonMarshalToString(v any) (string, error)

func JsonUnmarshalFromString 

func JsonUnmarshalFromString(data string, v any) error

func LoadJobRun 

func LoadJobRun(ctx Context, id uint64) (task, queue, status, payload string, err error)

LoadJobRun reads back one run's task, queue, status and payload. It exists so an application can re-dispatch a finished run without having to know how the columns are named.

func MarkJobRunFinished 

func MarkJobRunFinished(ctx Context, id uint64, res JobRunResult) error

MarkJobRunFinished records how an attempt ended. A pending status hands the run back for another attempt and leaves FinishedAt unset; anything outside the terminal set plus pending is rejected rather than written, because a status this table does not model would break the purge and the backlog silently.

func MarkJobRunStarted 

func MarkJobRunStarted(ctx Context, id uint64, attempt uint8) error

MarkJobRunStarted flips the row to running for the given attempt.

func NewConsumer 

func NewConsumer[B any](engine Engine, ref ConsumerRef[B]) *B

NewConsumer starts building the consumer for one declared name. It panics when the name is not registered, because that means the code and the declaration table disagree - a wiring bug, not a runtime condition.

func PurgeCDCOutbox 

func PurgeCDCOutbox(ctx Context, olderThan time.Duration, limit int) (int, error)

PurgeCDCOutbox deletes terminal rows older than olderThan and reports how many went. Pending rows are never touched however old they get: one of those is an undelivered event, and dropping it is the silent loss the outbox exists to prevent. A stuck relay therefore shows up as a growing backlog, not as vanished events.

func PurgeJobRuns 

func PurgeJobRuns(ctx Context, status string, olderThan time.Duration, limit int) (int, error)

PurgeJobRuns deletes rows of one terminal status older than olderThan and reports how many went. The status is a parameter, and only a terminal one is accepted, so no caller can reach a pending or running row: one of those is a job that has not finished, and deleting it destroys the only record that it was ever dispatched.

func RedispatchJobRun 

func RedispatchJobRun(ctx Context, runID uint64) (uint64, error)

RedispatchJobRun publishes a finished run's payload again as a brand new run, and returns the new run's ID.

This is the operator's retry, so it works from a task name rather than a Go type: an admin screen has the row, not the struct. The original row is left exactly as it was, because the attempt history is the reason the table is worth keeping.

An unfinished run is refused. Re-dispatching one would duplicate work that is still in flight, and JetStream is already redelivering it.

func RegisterAfterDeleteHandler 

func RegisterAfterDeleteHandler(engine Engine, cacheIndex string, handler func(Context, Entity) error)

func RegisterAfterInsertHandler 

func RegisterAfterInsertHandler(engine Engine, cacheIndex string, handler func(Context, Entity) error)

func RegisterAfterUpdateHandler 

func RegisterAfterUpdateHandler(engine Engine, cacheIndex string, handler func(Context, Entity, map[string]any) error)

func RegisterEntityPublisher 

func RegisterEntityPublisher[E any](
	table string,
	buildEvent func(entity *E, op DirtyOp, beforeOrigin map[string]any) ([]byte, error),
)

RegisterEntityPublisher records a per-entity publisher closure under the entity's reflect.Type. Called from generated `init()` blocks. The closure receives the entity + op + before-state and returns the marshaled wire payload - the generator emits the typed snapshot construction inline so the entity type and the JSON payload type stay independent.

The table name is the argument rather than the subject because the subject format is orm's to own; generated code names the table it already knows.

Duplicate registrations for the same E silently replace, so re-running tests with hot-reload doesn't error.

func SQLScanTarget 

func SQLScanTarget(destination any) any

SQLScanTarget adapts numeric destinations used by generated batch readers. The adapters share the destination itself, not a separately allocated wrapper. Successful text conversions avoid database/sql's temporary numeric strings; uncommon sources and failed conversions retain database/sql's behavior.

func ScanIDs added in v1.0.1 

func ScanIDs(rows Rows, ids []uint64) ([]uint64, error)

ScanIDs appends the single uint64 column of every remaining row to ids and returns the result. One scan target is shared by all rows, so a page of IDs costs a constant number of allocations instead of database/sql's per-row numeric-to-string conversion of int64 into uint64.

func UniqueIndexKeyHash 

func UniqueIndexKeyHash(values ...any) string

UniqueIndexKeyHash addresses a row by the values looked up. sha256 because the values are caller-supplied and fnv32a was grindable in seconds.

func UniqueIndexKeySegment 

func UniqueIndexKeySegment(indexName string, columns []string) string

UniqueIndexKeySegment names an index inside an entity's key space. Columns are folded in because the name alone is not unique: repointing "X on (Code)" to "X on (Slug)" would keep the old keys addressable.

Types

type Alter 

type Alter struct {
	SQL    string
	Pool   string
	Kind   AlterKind
	Table  string
	Entity string // empty for a table with no registered entity
	Safety AlterSafety
	Reason string
}

func GetAlters 

func GetAlters(ctx Context) (alters []Alter, err error)

func SplitAlters 

func SplitAlters(in []Alter) (safe, destructive []Alter)

SplitAlters partitions a plan into what may be applied during a rollout and what may not.

func (Alter) Exec 

func (a Alter) Exec(ctx Context) error

func (Alter) IsSafe 

func (a Alter) IsSafe() bool

type AlterKind 

type AlterKind string

AlterKind is what metrics label and listings sort by. 

const (
	AlterKindCreateTable    AlterKind = "create_table"
	AlterKindAddColumn      AlterKind = "add_column"
	AlterKindSetDefault     AlterKind = "set_default"
	AlterKindAddIndex       AlterKind = "add_index"
	AlterKindAddUniqueIndex AlterKind = "add_unique_index"
	AlterKindRebuildIndex   AlterKind = "rebuild_index"
	AlterKindChangeColumn   AlterKind = "change_column"
	AlterKindDropIndex      AlterKind = "drop_index"
	AlterKindDropColumn     AlterKind = "drop_column"
	AlterKindDropForeignKey AlterKind = "drop_foreign_key"
	AlterKindModifyTable    AlterKind = "modify_table"
	AlterKindConvertTable   AlterKind = "convert_table"
	AlterKindDropTable      AlterKind = "drop_table"
)

func AllAlterKinds 

func AllAlterKinds() []AlterKind

AllAlterKinds is every kind the diff can emit, for callers building a policy over them.

type AlterSafety 

type AlterSafety uint8

AlterSafety answers one question: can the previous code version keep running against this statement? AlterDestructive is the zero value so an unclassified path fails closed. 

const (
	// AlterDestructive must not run while the previous version is live.
	AlterDestructive AlterSafety = iota
	// AlterSafe is forward-compatible with the previous code version.
	AlterSafe
)

func (AlterSafety) String 

func (s AlterSafety) String() string

type BaseWhere 

type BaseWhere struct {
	// contains filtered or unexported fields
}

func NewWhere 

func NewWhere(query string, parameters ...any) *BaseWhere

func (*BaseWhere) Append 

func (w *BaseWhere) Append(query string, parameters ...any)

func (*BaseWhere) GetParameters 

func (w *BaseWhere) GetParameters() []any

func (*BaseWhere) IsWithFakeDeletes 

func (w *BaseWhere) IsWithFakeDeletes() bool

func (*BaseWhere) SetParameter 

func (w *BaseWhere) SetParameter(index int, param any) *BaseWhere

func (*BaseWhere) SetParameters 

func (w *BaseWhere) SetParameters(params ...any) *BaseWhere

func (*BaseWhere) String 

func (w *BaseWhere) String() string

func (*BaseWhere) WithFakeDeletes 

func (w *BaseWhere) WithFakeDeletes() *BaseWhere

type BatchDispatch 

type BatchDispatch func(ctx Context, msgs []*NatsMessage) error

BatchDispatch handles every message for one entity within a single fetched batch. Returning nil acks the whole group; returning err leaves the whole group unacked for redelivery, so batch handlers must be idempotent.

func BuildBatchDispatch 

func BuildBatchDispatch[T any](
	handler func(ctx Context, evs []*DirtyEvent[T]) error, opts ...HandlerOption,
) BatchDispatch

BuildBatchDispatch wraps a typed user batch handler into the generic BatchDispatch closure. Generated OnXBatch methods call this.

Every message in the group is decoded up front and filtered through WatchFields; if nothing survives the filter the handler is skipped and the group acks. A decode failure fails the whole group rather than silently dropping a message.

type BoolField 

type BoolField struct{ Column string }

BoolField represents a boolean column.

func (BoolField) ColumnName 

func (f BoolField) ColumnName() string

func (BoolField) Is 

func (f BoolField) Is(v bool) Condition

type CDCOutboxEntity 

type CDCOutboxEntity struct {
	ID           uint64     `orm:"table=cdc_outbox"`
	Status       string     `orm:"enum=pending,dispatched,stored;enumName=CDCOutboxStatus;required"`
	EntityName   string     `orm:"required;length=100"`
	EntityID     uint64     `orm:"required"`
	Op           uint8      `orm:"required"`
	Payload      string     `orm:"length=max;required"`
	DispatchedAt *time.Time `orm:"time"`
	CreatedAt    time.Time  `orm:"time"`
}

CDCOutboxEntity is the transactional outbox table. It is declared here rather than by the application so the column set is a compile-time fact on both sides: orm writes it with raw SQL, the application reads it through the generated provider. Register it with `registry.RegisterEntity` and it generates, alters and queries like any other entity.

An entity tagged `orm:"outbox"` gets one row per write, inside that write's own transaction. With `cdc` the row is a delivery guarantee; alone it is a change log. See the package docs for the full truth table.

func (CDCOutboxEntity) Indexes 

func (e CDCOutboxEntity) Indexes() [][]string

Indexes covers both readers with one index: the relay scans pending rows past a grace period, the purge scans terminal rows past a retention window.

type CDCOutboxRelayResult 

type CDCOutboxRelayResult struct {
	Fetched    int
	Dispatched int
	Published  map[string]int
}

CDCOutboxRelayResult reports one page of relay work.

Dispatched is the count of rows actually marked delivered, and it is what a paging caller must loop on. Fetched only says how many rows were read: a page that fails to publish stays pending, so paging on Fetched would re-read the same rows forever. Published is the message count per NATS pool, which is now one per row however many consumers read that entity, and exists for a labelled metric rather than for control flow.

func RelayCDCOutbox 

func RelayCDCOutbox(ctx Context, minAge time.Duration, limit int) (CDCOutboxRelayResult, error)

RelayCDCOutbox republishes one page of outbox rows whose inline publish never landed, then marks the ones that made it onto the wire.

It lives here rather than in the application because the write side lives here: the two halves have to agree on the status values, the column meanings and above all the deterministic Nats-Msg-Id that JetStream dedups on. An application-side relay that drifted on any of those would turn every recovery into a duplicate delivery instead of a no-op.

minAge skips rows still racing the ORM's own post-commit mark, and must stay well under the stream's DuplicateWindow (10m by default) so a republish is still deduped. Publish happens before the mark, so delivery is at-least-once: a row is never recorded as delivered before it is.

A pool whose batch fails leaves its rows pending for the next call and contributes to the returned error; other pools still publish and mark. The result is populated either way.

type Clickhouse 

type Clickhouse interface {
	GetConfig() ClickhouseConfig
	GetDBClient() DBClient
	SetMockDBClient(mock DBClient)
	Exec(ctx Context, query string, args ...any) (ExecResult, error)
	QueryRow(ctx Context, query Where, toFill ...any) (found bool, err error)
	Query(ctx Context, query string, args ...any) (rows Rows, close func(), err error)
}

type ClickhouseAlter 

type ClickhouseAlter struct {
	SQL    string
	Pool   string
	Kind   AlterKind
	Safety AlterSafety
}

ClickhouseAlter holds a pending ClickHouse DDL operation.

func GetClickhouseAlters 

func GetClickhouseAlters(ctx Context) ([]ClickhouseAlter, error)

GetClickhouseAlters compares registered ClickHouse table definitions with actual database schemas and returns the DDL statements needed to synchronize them.

func (ClickhouseAlter) Exec 

func (a ClickhouseAlter) Exec(ctx Context) error

Exec executes the ClickHouse DDL statement.

func (ClickhouseAlter) IsSafe 

func (a ClickhouseAlter) IsSafe() bool

type ClickhouseColumnOptions 

type ClickhouseColumnOptions struct {
	Default      string // DEFAULT expression
	Materialized string // MATERIALIZED expression
	Alias        string // ALIAS expression
	Codec        string // e.g. "ZSTD(1)"
	TTL          string // column-level TTL expression
	Comment      string
}

ClickhouseColumnOptions holds optional settings for a ClickHouse column.

type ClickhouseConfig 

type ClickhouseConfig interface {
	GetCode() string
	GetDataSourceURI() string
	GetDatabaseName() string
	GetOptions() *ClickhouseOptions
	// contains filtered or unexported methods
}

type ClickhouseOptions 

type ClickhouseOptions struct {
	ConnMaxLifetime    time.Duration
	MaxOpenConnections int
	MaxIdleConnections int
	IgnoredTables      []string
	// Replicated turns MergeTree-family engines of this pool's tables into their Replicated* form (a Replicated database).
	Replicated bool
}

type ClickhouseTableBuilder 

type ClickhouseTableBuilder struct {
	// contains filtered or unexported fields
}

ClickhouseTableBuilder defines a ClickHouse table schema using a fluent API.

func NewClickhouseTable 

func NewClickhouseTable(tableName, poolCode string) *ClickhouseTableBuilder

NewClickhouseTable creates a new ClickHouse table builder.

func (*ClickhouseTableBuilder) Column 

func (b *ClickhouseTableBuilder) Column(name, typeName string) *ClickhouseTableBuilder

Column adds a simple column.

func (*ClickhouseTableBuilder) ColumnAlias 

func (b *ClickhouseTableBuilder) ColumnAlias(name, typeName, expr string) *ClickhouseTableBuilder

ColumnAlias adds a column with an ALIAS expression.

func (*ClickhouseTableBuilder) ColumnCodec 

func (b *ClickhouseTableBuilder) ColumnCodec(name, typeName, codec string) *ClickhouseTableBuilder

ColumnCodec adds a column with a compression CODEC.

func (*ClickhouseTableBuilder) ColumnComment 

func (b *ClickhouseTableBuilder) ColumnComment(name, typeName, comment string) *ClickhouseTableBuilder

ColumnComment adds a column with a comment.

func (*ClickhouseTableBuilder) ColumnDefault 

func (b *ClickhouseTableBuilder) ColumnDefault(name, typeName, defaultExpr string) *ClickhouseTableBuilder

ColumnDefault adds a column with a DEFAULT expression.

func (*ClickhouseTableBuilder) ColumnFull 

func (b *ClickhouseTableBuilder) ColumnFull(name, typeName string, opts ClickhouseColumnOptions) *ClickhouseTableBuilder

ColumnFull adds a column with all possible options.

func (*ClickhouseTableBuilder) ColumnMaterialized 

func (b *ClickhouseTableBuilder) ColumnMaterialized(name, typeName, expr string) *ClickhouseTableBuilder

ColumnMaterialized adds a column with a MATERIALIZED expression.

func (*ClickhouseTableBuilder) ColumnTTL 

func (b *ClickhouseTableBuilder) ColumnTTL(name, typeName, ttl string) *ClickhouseTableBuilder

ColumnTTL adds a column with a column-level TTL expression.

func (*ClickhouseTableBuilder) Comment 

func (b *ClickhouseTableBuilder) Comment(comment string) *ClickhouseTableBuilder

Comment sets the table comment.

func (*ClickhouseTableBuilder) Engine 

func (b *ClickhouseTableBuilder) Engine(engine string) *ClickhouseTableBuilder

Engine sets the table engine (e.g. "MergeTree", "ReplacingMergeTree(Version)").

func (*ClickhouseTableBuilder) OrderBy 

func (b *ClickhouseTableBuilder) OrderBy(columns ...string) *ClickhouseTableBuilder

OrderBy sets the ORDER BY columns.

func (*ClickhouseTableBuilder) PartitionBy 

func (b *ClickhouseTableBuilder) PartitionBy(expr string) *ClickhouseTableBuilder

PartitionBy sets the PARTITION BY expression.

func (*ClickhouseTableBuilder) PrimaryKey 

func (b *ClickhouseTableBuilder) PrimaryKey(columns ...string) *ClickhouseTableBuilder

PrimaryKey sets the PRIMARY KEY columns. If not set, defaults to ORDER BY.

func (*ClickhouseTableBuilder) Setting 

func (b *ClickhouseTableBuilder) Setting(key, value string) *ClickhouseTableBuilder

Setting adds a table-level SETTINGS key=value pair.

func (*ClickhouseTableBuilder) TTL 

func (b *ClickhouseTableBuilder) TTL(expr string) *ClickhouseTableBuilder

TTL sets the table-level TTL expression.

type ColumnSchemaDefinition 

type ColumnSchemaDefinition struct {
	ColumnName string
	Definition string
}

type Condition 

type Condition interface {
	ToSQL() (clause string, params []any)

	Not() Condition
	// contains filtered or unexported methods
}

Condition represents a single WHERE clause predicate.

type Config 

type Config struct {
	MySQlPools         []ConfigMysql         `yaml:"mysqlPools"`
	RedisPools         []ConfigRedis         `yaml:"redisPools"`
	RedisSentinelPools []ConfigRedisSentinel `yaml:"redisSentinelPools"`
	ClickhousePools    []ConfigClickhouse    `yaml:"clickhousePools"`
	NatsPools          []ConfigNats          `yaml:"natsPools"`
}

type ConfigClickhouse 

type ConfigClickhouse struct {
	Code               string   `yaml:"code" validate:"required"`
	URI                string   `yaml:"uri" validate:"required"`
	ConnMaxLifetime    int      `yaml:"connMaxLifetime"`
	MaxOpenConnections int      `yaml:"maxOpenConnections"`
	MaxIdleConnections int      `yaml:"maxIdleConnections"`
	IgnoredTables      []string `yaml:"ignoredTables"`
	Replicated         bool     `yaml:"replicated"`
}

type ConfigMysql 

type ConfigMysql struct {
	Code               string   `yaml:"code" validate:"required"`
	URI                string   `yaml:"uri" validate:"required"`
	ConnMaxLifetime    int      `yaml:"connMaxLifetime"`
	MaxOpenConnections int      `yaml:"maxOpenConnections"`
	MaxIdleConnections int      `yaml:"maxIdleConnections"`
	DefaultEncoding    string   `yaml:"defaultEncoding"`
	DefaultCollate     string   `yaml:"defaultCollate"`
	IgnoredTables      []string `yaml:"ignoredTables"`
}

type ConfigNats 

type ConfigNats struct {
	Code                 string               `yaml:"code" validate:"required"`
	URLs                 []string             `yaml:"urls" validate:"required"`
	ClientID             string               `yaml:"clientID"`
	MaxReconnects        int                  `yaml:"maxReconnects"`
	ReconnectWaitMs      int                  `yaml:"reconnectWaitMs"`
	ReconnectBufSize     int                  `yaml:"reconnectBufSize"`
	ConnectTimeoutMs     int                  `yaml:"connectTimeoutMs"`
	RetryOnFailedConnect bool                 `yaml:"retryOnFailedConnect"`
	AuthToken            string               `yaml:"authToken"`
	AuthUser             string               `yaml:"authUser"`
	AuthPassword         string               `yaml:"authPassword"`
	AuthCredsFile        string               `yaml:"authCredsFile"`
	AuthNKeySeed         string               `yaml:"authNKeySeed"`
	IgnoredSubjects      []string             `yaml:"ignoredSubjects"`
	IgnoredConsumers     []string             `yaml:"ignoredConsumers"`
	Consumers            []ConfigNatsConsumer `yaml:"consumers"`
	Streams              []ConfigNatsStream   `yaml:"streams"`
}

type ConfigNatsConsumer 

type ConfigNatsConsumer struct {
	Name           string   `yaml:"name" validate:"required"`
	FilterSubjects []string `yaml:"filterSubjects"`
	AckWaitMs      int      `yaml:"ackWaitMs"`
	MaxAckPending  int      `yaml:"maxAckPending"`
	MaxDeliver     int      `yaml:"maxDeliver"`
}

type ConfigNatsStream 

type ConfigNatsStream struct {
	Name            string   `yaml:"name" validate:"required"`
	Subjects        []string `yaml:"subjects" validate:"required"`
	MaxAgeMs        int      `yaml:"maxAgeMs"`
	MaxBytes        int64    `yaml:"maxBytes"`
	MaxMsgSize      int32    `yaml:"maxMsgSize"`
	Replicas        int      `yaml:"replicas"`
	DuplicateWindow int      `yaml:"duplicateWindowMs"`
	Storage         string   `yaml:"storage"`   // "file" or "memory"
	Retention       string   `yaml:"retention"` // "limits", "interest", "workqueue"
}

type ConfigRedis 

type ConfigRedis struct {
	Code     string `yaml:"code" validate:"required"`
	URI      string `yaml:"uri" validate:"required"`
	Database int    `yaml:"database"`
	User     string `yaml:"user"`
	Password string `yaml:"password"`
}

type ConfigRedisSentinel 

type ConfigRedisSentinel struct {
	Code       string   `yaml:"code" validate:"required"`
	MasterName string   `yaml:"masterName" validate:"required"`
	Database   int      `yaml:"database"`
	Sentinels  []string `yaml:"sentinels"`
	User       string   `yaml:"user"`
	Password   string   `yaml:"password"`
}

type ConsumerBuilder 

type ConsumerBuilder struct {
	// contains filtered or unexported fields
}

ConsumerBuilder is the shared state every generated builder embeds. Exported because generated code in the application's package calls AddDispatch.

func (*ConsumerBuilder) AddBatchDispatch 

func (b *ConsumerBuilder) AddBatchDispatch(entityName string, d BatchDispatch)

AddBatchDispatch registers a typed batch dispatch closure. Called from generated OnXBatch methods. Not for direct use.

func (*ConsumerBuilder) AddDispatch 

func (b *ConsumerBuilder) AddDispatch(key string, d Dispatch)

AddDispatch registers a typed dispatch closure under a dispatch key - an entity's Go name, or a task's name. Called from generated OnX methods. Not for direct use.

func (*ConsumerBuilder) Build 

func (b *ConsumerBuilder) Build() StreamConsumer

Build wraps the accumulated dispatch table into a StreamConsumer ready to be driven by the application's own loop.

Panics if an entity has both a per-message and a batch handler: that is a wiring mistake, and silently preferring one would hide a dropped handler.

func (*ConsumerBuilder) Name 

func (b *ConsumerBuilder) Name() ConsumerName

Name is the consumer this builder is wiring.

type ConsumerDef 

type ConsumerDef struct {
	// Name is the durable name and the metrics label. Must match
	// ^[a-z][a-z0-9-]*$.
	Name ConsumerName

	// Entities are the entity structs whose changes this consumer wants, given
	// as the same values passed to RegisterEntity. Every one must be tagged
	// `orm:"cdc"`.
	//
	// Mutually exclusive with Queues: a JetStream consumer belongs to one
	// stream, and entity changes and tasks are on different ones.
	Entities []any

	// Queues are the task queues this consumer drains. A queue is the unit of
	// consumption - exactly one consumer may drain it - so listing individual
	// tasks would be a second way to say the same thing.
	Queues []Queue

	// AckWait must exceed the slowest handler on this consumer, or JetStream
	// redelivers work that is still running.
	AckWait       time.Duration
	MaxAckPending int
	// MaxDeliver defaults to unlimited: an entity event that keeps failing is a
	// bug to fix, and dropping it silently loses a change no reindex knows about.
	MaxDeliver int
	NatsPool   string
}

ConsumerDef declares one consumer: its name, and which subjects it wants.

The direction matters. An entity publishes its own subject and knows nothing about who reads it; the reader declares what it needs. That inversion is why adding a consumer for `orders` is a change in one place instead of an edit to the orders entity tag.

type ConsumerHandle 

type ConsumerHandle interface {
	Name() ConsumerName
	Subjects() []Subject
	Enqueue(orm Context, entities []Entity) error
}

ConsumerHandle is the untyped view of a generated consumer ref, for code that holds consumers in a table and does not know their builder types.

type ConsumerName 

type ConsumerName string

ConsumerName names a declared consumer. It doubles as the JetStream durable name, so the two can never drift and an operator reading `nats consumer ls` sees the same names the code declares.

type ConsumerRef 

type ConsumerRef[B any] struct {
	// contains filtered or unexported fields
}

ConsumerRef ties a declared consumer to its generated typed builder. `B` is the generated builder type, and inference at NewConsumer's call site resolves the right typed return.

Constructed by generated code; application code references the generated value (e.g. gen.ConsumerOrderIndexer) and never instantiates this.

func NewConsumerRef 

func NewConsumerRef[B any](
	name ConsumerName, subjects []Subject, newBuilder func(core *ConsumerBuilder) *B,
) ConsumerRef[B]

NewConsumerRef is the generator-facing factory.

func (ConsumerRef[B]) Enqueue 

func (r ConsumerRef[B]) Enqueue(orm Context, entities []Entity) error

Enqueue publishes a synthetic DirtyUpdate event for each entity onto this consumer's private replay subject, without any database mutation. Use it to re-drive one consumer over rows that have not otherwise changed.

The event mirrors the entity's currently-loaded state for both Before and After, so a handler using WatchFields correctly observes no change and skips it - only handlers without WatchFields (full reindexers) act on a replay.

Every entity is published in one batch, so replaying a page of rows costs one round-trip rather than one per row.

func (ConsumerRef[B]) Name 

func (r ConsumerRef[B]) Name() ConsumerName

func (ConsumerRef[B]) Subjects 

func (r ConsumerRef[B]) Subjects() []Subject

type Context 

type Context interface {
	Context() context.Context
	Clone() Context
	CloneWithContext(context context.Context) Context
	Engine() Engine
	DisableContextCache()
	Save(entities ...Entity) error
	Reload(entities ...Entity) error
	Delete(entities ...Entity) error
	ForceDelete(entities ...Entity) error
	Transaction(fn func(tx Context) error) error
	InTransaction() bool
	DB(pool string) DBBase
	InvalidateCacheKey(pool, key string)
	RedisPipeLine(pool string) *RedisPipeLine
	DatabasePipeLine(pool string) *DatabasePipeline
	RegisterQueryLogger(handler LogHandler, options QueryLoggerOptions)
	EnableQueryDebug()
	EnableQueryDebugCustom(options QueryLoggerOptions)
	SetMetaData(key, value string)
	GetMetaData() Meta

	GetFromContextCache(cacheIndex string, id uint64) Entity
	SetInContextCache(cacheIndex string, id uint64, entity Entity)
	// contains filtered or unexported methods
}

func PrepareTables 

func PrepareTables(t *testing.T, registry Registry, entities ...any) (orm Context)

func PrepareTablesWithConsumers 

func PrepareTablesWithConsumers(
	t *testing.T, registry Registry, consumers []ConsumerDef, entities ...any,
) (orm Context)

PrepareTablesWithConsumers sets up MySQL/Redis/NATS, registers the given consumers, and purges both streams so test runs are independent.

Production callers register consumers from their own declaration table; this helper exists so tests don't have to mirror that boilerplate.

func PrepareTablesWithNats 

func PrepareTablesWithNats(t *testing.T, registry Registry, entities ...any) (orm Context)

type DB 

type DB interface {
	DBBase
	Begin(ctx Context) (DBTransaction, error)
}

type DBBase 

type DBBase interface {
	GetConfig() MySQLConfig
	GetDBClient() DBClient
	SetMockDBClient(mock DBClient)
	Exec(ctx Context, query string, args ...any) (ExecResult, error)
	QueryRow(ctx Context, query Where, toFill ...any) (found bool, err error)
	Query(ctx Context, query string, args ...any) (rows Rows, close func(), err error)
}

type DBClient 

type DBClient interface {
	DBClientQuery
}

type DBClientNoTX 

type DBClientNoTX interface {
	DBClientQuery
	Begin() (*sql.Tx, error)
}

type DBClientQuery 

type DBClientQuery interface {
	Exec(query string, args ...any) (sql.Result, error)
	ExecContext(context context.Context, query string, args ...any) (sql.Result, error)
	QueryRow(query string, args ...any) *sql.Row
	QueryRowContext(context context.Context, query string, args ...any) *sql.Row
	Query(query string, args ...any) (*sql.Rows, error)
	QueryContext(context context.Context, query string, args ...any) (*sql.Rows, error)
}

type DBQuery 

type DBQuery struct {
	// contains filtered or unexported fields
}

DBQuery is a type-safe query builder for MySQL queries.

func NewQuery 

func NewQuery() *DBQuery

NewQuery creates a new empty DBQuery.

func (*DBQuery) BuildLimitClause 

func (q *DBQuery) BuildLimitClause() string

BuildLimitClause returns "LIMIT offset,count" or empty string.

func (*DBQuery) BuildOrderClause 

func (q *DBQuery) BuildOrderClause() string

BuildOrderClause returns "ORDER BY `col1` ASC, `col2` DESC" or empty string.

func (*DBQuery) BuildWhereClause 

func (q *DBQuery) BuildWhereClause() (string, []any)

BuildWhereClause builds the WHERE portion (without "WHERE" keyword). Returns empty string and nil params if no conditions.

func (*DBQuery) Filter 

func (q *DBQuery) Filter(conditions ...Condition) *DBQuery

Filter adds typed conditions to the query.

func (*DBQuery) FilterWhere 

func (q *DBQuery) FilterWhere(w Where) *DBQuery

FilterWhere adds a raw Where clause to the query.

func (*DBQuery) GetConditions 

func (q *DBQuery) GetConditions() []Condition

GetConditions returns typed conditions for generated code's unique index detection.

func (*DBQuery) IsWithFakeDeletes 

func (q *DBQuery) IsWithFakeDeletes() bool

IsWithFakeDeletes returns true if rawWhere has WithFakeDeletes set.

func (*DBQuery) PageSize added in v1.0.1 

func (q *DBQuery) PageSize() int

PageSize returns the pager's page size, or 0 when the query has no pager.

func (*DBQuery) Pager 

func (q *DBQuery) Pager(p *Pager) *DBQuery

Pager sets the pagination for the query.

func (*DBQuery) SortByASC 

func (q *DBQuery) SortByASC(f Field) *DBQuery

SortByASC adds an ascending sort clause.

func (*DBQuery) SortByDESC 

func (q *DBQuery) SortByDESC(f Field) *DBQuery

SortByDESC adds a descending sort clause.

func (*DBQuery) WriteSelect added in v1.0.1 

func (q *DBQuery) WriteSelect(b *strings.Builder, prefix string, excludeFakeDeleted, withOrder, withLimit bool) []any

WriteSelect appends the query to b after prefix: the WHERE clause built from the typed conditions and the raw Where, then the ORDER BY clause when withOrder is set and the pager's LIMIT clause when withLimit is set. excludeFakeDeleted adds the `FakeDelete` = 0 guard unless the raw Where allows fake-deleted rows. It returns the bound parameters in SQL order. The builder grows once up front, so a typical query costs one string and one parameter slice.

type DBTransaction 

type DBTransaction interface {
	DBBase
	Commit(ctx Context) error
	Rollback(ctx Context) error
}

type DatabasePipeline 

type DatabasePipeline struct {
	// contains filtered or unexported fields
}

func (*DatabasePipeline) AddQuery 

func (dp *DatabasePipeline) AddQuery(query string, parameters ...any)

func (*DatabasePipeline) AddQueryForTable 

func (dp *DatabasePipeline) AddQueryForTable(table, query string, parameters ...any)

func (*DatabasePipeline) Exec 

func (dp *DatabasePipeline) Exec(ctx Context) error

type DirtyEvent 

type DirtyEvent[T any] struct {
	Op     DirtyOp `json:"op"`
	ID     uint64  `json:"id"`
	Before *T      `json:"before,omitempty"` // nil for Insert
	After  *T      `json:"after,omitempty"`  // nil for Delete
	TsMs   int64   `json:"ts_ms"`
}

DirtyEvent is the typed change envelope published when an entity tagged `orm:"cdc"` is flushed. The generator emits per-entity aliases: 

type OrdersDirtyEvent = orm.DirtyEvent[Orders]

type DirtyOp 

type DirtyOp uint8

DirtyOp identifies the kind of database mutation that produced an event. 

const (
	DirtyInsert DirtyOp = 1
	DirtyUpdate DirtyOp = 2
	DirtyDelete DirtyOp = 3
)

func (DirtyOp) String 

func (op DirtyOp) String() string

String returns the lowercase name (used in logs / wire header).

type Dispatch 

type Dispatch func(ctx Context, msg *NatsMessage) error

Dispatch is a typed handler after type erasure. Generated code produces these through BuildDispatch; application code does not write them.

func BuildDispatch 

func BuildDispatch[T any](handler func(ctx Context, ev *DirtyEvent[T]) error, opts ...HandlerOption) Dispatch

BuildDispatch wraps a typed user handler into the generic Dispatch closure. Generated OnX methods call this.

func BuildTaskDispatch 

func BuildTaskDispatch[T any](handler func(ctx Context, task *T) error) Dispatch

BuildTaskDispatch erases a typed task handler's payload type. A decode failure comes back wrapped in ErrPayloadUndecodable so the consumer can tell a bad message from a failing handler and dead-letter it immediately.

type DispatchOption 

type DispatchOption func(*dispatchConfig)

func WithIdempotencyKey 

func WithIdempotencyKey(key string) DispatchOption

WithIdempotencyKey collapses repeat dispatches of the same logical work inside the task stream's dedup window. The collapsed dispatch's run row is closed as deduplicated, so it is visible rather than pending forever.

type Engine 

type Engine interface {
	NewContext(parent context.Context) Context
	DB(code string) DB
	Clickhouse(code string) Clickhouse
	Nats(code string) Nats
	Redis(code string) RedisCache
	Registry() EngineRegistry
	Option(key string) any
	NextID() uint64
	SetNodeID(node int64)
}

type EngineRegistry 

type EngineRegistry interface {
	DBPools() map[string]DB
	ClickhousePools() map[string]Clickhouse
	NatsPools() map[string]Nats
	RedisPools() map[string]RedisCache
	Option(key string) any
	// contains filtered or unexported methods
}

type EngineSetter 

type EngineSetter interface {
	SetOption(key string, value any)
}

type Entity 

type Entity interface {
	PrivateFlush() error
	PrivateFlushed()
	PrivateFlushEvent() (uint8, map[string]any)
	PrivateGetDatabaseBind() map[string]any
	PrivateIsNew() bool
	PrivateReload() (found bool, err error)
	GetID() uint64
}

type EntityCachedUniqueIndexes 

type EntityCachedUniqueIndexes interface {
	CachedUniqueIndexes() [][]string
}

type EntityIndexes 

type EntityIndexes interface {
	Indexes() [][]string
}

type EntityProvider 

type EntityProvider interface {
	TableName() string
	DBCode() string
}

EntityProvider is the base interface implemented by all generated entity providers.

type EntityStreamOptions 

type EntityStreamOptions struct {
	Storage  jetstream.StorageType
	Replicas int
	MaxAge   time.Duration
	// MaxBytes bounds the stream. Worth setting explicitly: replay shares this
	// stream, so an unbounded reindex either evicts live events or fills the
	// disk, and the second failure takes the publishers down with it.
	MaxBytes        int64
	DuplicateWindow time.Duration
	NatsPool        string
}

EntityStreamOptions tunes the single stream that carries entity changes and replays. Registering it is optional; the stream exists with these defaults either way.

type EntityUniqueIndexes 

type EntityUniqueIndexes interface {
	UniqueIndexes() [][]string
}

type EnumField 

type EnumField struct{ Column string }

EnumField represents an enum column stored as a string.

func (EnumField) ColumnName 

func (f EnumField) ColumnName() string

func (EnumField) In 

func (f EnumField) In(values ...any) Condition

func (EnumField) Is 

func (f EnumField) Is(v any) Condition

type EqCondition 

type EqCondition interface {
	Condition
	ColumnName() string
	EqValue() any
}

EqCondition is exposed for generated code to type-assert and extract values for cached unique index detection.

type ExecResult 

type ExecResult interface {
	LastInsertId() (uint64, error)
	RowsAffected() (uint64, error)
}

type Field 

type Field interface {
	ColumnName() string
}

Field is implemented by all field types for use in SortByASC/SortByDESC.

type FloatField 

type FloatField struct{ Column string }

FloatField represents a floating-point column.

func (FloatField) ColumnName 

func (f FloatField) ColumnName() string

func (FloatField) Eq 

func (f FloatField) Eq(v float64) Condition

func (FloatField) Gt 

func (f FloatField) Gt(v float64) Condition

func (FloatField) Gte 

func (f FloatField) Gte(v float64) Condition

func (FloatField) In 

func (f FloatField) In(values ...float64) Condition

func (FloatField) Lt 

func (f FloatField) Lt(v float64) Condition

func (FloatField) Lte 

func (f FloatField) Lte(v float64) Condition

type HandlerOption 

type HandlerOption func(*handlerConfig)

HandlerOption configures a single typed handler (field filtering, etc).

func WatchFields 

func WatchFields(fields ...Field) HandlerOption

WatchFields skips Update events where none of the listed entity fields differ between Before and After. Insert and Delete events always fire. The Field type is the same one providers expose (e.g. gen.ShowsProvider.Fields.Status), so callers don't need column-name strings.

A replayed event carries identical Before and After, so a handler using WatchFields deliberately ignores replays - only full reindexers, which do not filter, act on them.

type IndexSchemaDefinition 

type IndexSchemaDefinition struct {
	Name       string
	Unique     bool
	Duplicated bool
	// contains filtered or unexported fields
}

func (*IndexSchemaDefinition) GetColumns 

func (ti *IndexSchemaDefinition) GetColumns() []string

func (*IndexSchemaDefinition) SetColumns 

func (ti *IndexSchemaDefinition) SetColumns(columns []string)

type IntField 

type IntField struct{ Column string }

IntField represents a signed integer column.

func (IntField) ColumnName 

func (f IntField) ColumnName() string

func (IntField) Eq 

func (f IntField) Eq(v int64) Condition

func (IntField) Gt 

func (f IntField) Gt(v int64) Condition

func (IntField) Gte 

func (f IntField) Gte(v int64) Condition

func (IntField) In 

func (f IntField) In(values ...int64) Condition

func (IntField) Lt 

func (f IntField) Lt(v int64) Condition

func (IntField) Lte 

func (f IntField) Lte(v int64) Condition

type JobRunEntity 

type JobRunEntity struct {
	ID         uint64     `orm:"table=job_runs"`
	Task       string     `orm:"required;length=100"`
	Queue      string     `orm:"required;length=50"`
	Status     string     `orm:"enum=pending,running,succeeded,failed,deduplicated;enumName=JobRunStatus;required"`
	Payload    string     `orm:"length=max;required"`
	Attempts   uint8      `orm:"required"`
	LastError  string     `orm:"length=max;required"`
	DurationMs uint32     `orm:"required"`
	StartedAt  *time.Time `orm:"time"`
	FinishedAt *time.Time `orm:"time"`
	CreatedAt  time.Time  `orm:"time"`
}

JobRunEntity records one dispatch of one task: its payload, how many attempts it took, and how it ended. The application registers it like any entity and gets the table, the provider and the status enum generated from it.

orm owns this struct because orm owns the transitions. An application that hand-wrote the same table would have to keep its column names and status values in step with the consumer's ladder.

func (JobRunEntity) Indexes 

func (e JobRunEntity) Indexes() [][]string

Status+CreatedAt serves the purge and the backlog; Task+CreatedAt serves the admin screen's per-task history.

type JobRunResult 

type JobRunResult struct {
	Status   string
	Err      string
	Duration time.Duration
}

JobRunResult is how an attempt ended. Status is one of the terminal statuses, or pending to hand the run back for another attempt.

type JobRunSpec 

type JobRunSpec struct {
	Task    string
	Queue   string
	Payload string
}

JobRunSpec is the row a dispatch creates before it publishes.

type Lock 

type Lock struct {
	// contains filtered or unexported fields
}

func (*Lock) Refresh 

func (l *Lock) Refresh(ctx Context, ttl time.Duration) (bool, error)

func (*Lock) Release 

func (l *Lock) Release(ctx Context)

func (*Lock) TTL 

func (l *Lock) TTL(ctx Context) (time.Duration, error)

type Locker 

type Locker struct {
	// contains filtered or unexported fields
}

func (*Locker) Obtain 

func (l *Locker) Obtain(ctx Context, key string, ttl time.Duration, waitTimeout time.Duration) (lock *Lock, obtained bool, err error)

type LogHandler 

type LogHandler interface {
	Handle(ctx Context, log map[string]any)
}

type Meta 

type Meta map[string]string

func (Meta) Get 

func (m Meta) Get(key string) string

type MockDBClient 

type MockDBClient struct {
	OriginDB            DBClient
	PrepareMock         func(query string) (*sql.Stmt, error)
	ExecMock            func(query string, args ...any) (sql.Result, error)
	ExecContextMock     func(context context.Context, query string, args ...any) (sql.Result, error)
	QueryRowMock        func(query string, args ...any) *sql.Row
	QueryRowContextMock func(context context.Context, query string, args ...any) *sql.Row
	QueryMock           func(query string, args ...any) (*sql.Rows, error)
	QueryContextMock    func(context context.Context, query string, args ...any) (*sql.Rows, error)
	BeginMock           func() (*sql.Tx, error)
	CommitMock          func() error
	RollbackMock        func() error
}

func (*MockDBClient) Begin 

func (m *MockDBClient) Begin() (*sql.Tx, error)

func (*MockDBClient) Exec 

func (m *MockDBClient) Exec(query string, args ...any) (sql.Result, error)

func (*MockDBClient) ExecContext 

func (m *MockDBClient) ExecContext(context context.Context, query string, args ...any) (sql.Result, error)

func (*MockDBClient) Query 

func (m *MockDBClient) Query(query string, args ...any) (*sql.Rows, error)

func (*MockDBClient) QueryContext 

func (m *MockDBClient) QueryContext(context context.Context, query string, args ...any) (*sql.Rows, error)

func (*MockDBClient) QueryRow 

func (m *MockDBClient) QueryRow(query string, args ...any) *sql.Row

func (*MockDBClient) QueryRowContext 

func (m *MockDBClient) QueryRowContext(context context.Context, query string, args ...any) *sql.Row

type MockLogHandler 

type MockLogHandler struct {
	Logs []map[string]any
}

func (*MockLogHandler) Clear 

func (h *MockLogHandler) Clear()

func (*MockLogHandler) Handle 

func (h *MockLogHandler) Handle(_ Context, log map[string]any)

type MySQLConfig 

type MySQLConfig interface {
	GetCode() string
	GetDatabaseName() string
	GetDataSourceURI() string
	GetOptions() *MySQLOptions
	// contains filtered or unexported methods
}

type MySQLOptions 

type MySQLOptions struct {
	ConnMaxLifetime    time.Duration
	MaxOpenConnections int
	MaxIdleConnections int
	DefaultEncoding    string
	DefaultCollate     string
	IgnoredTables      []string
	// ServerSidePreparedStatements keeps go-sql-driver's default of preparing
	// every parametrized query on the server. By default the ORM adds
	// interpolateParams=true to the DSN (unless the DSN sets it), so a query
	// with parameters costs one round trip instead of prepare and execute.
	// The driver refuses interpolation for the multibyte connection charsets
	// big5, cp932, gb2312, gbk and sjis; set this for such DSNs.
	ServerSidePreparedStatements bool
}

type Nats 

type Nats interface {
	GetCode() string
	GetURLs() []string
	GetPoolOptions() *NatsPoolOptions
	Ping() error
	Publish(ctx Context, msg *NatsMessage) error
	PublishWithAck(ctx Context, msg *NatsMessage) (NatsPubAck, error)
	PublishBatch(ctx Context, msgs []*NatsMessage) error
	PublishAsync(ctx Context, msg *NatsMessage, callback func(*NatsMessage, error))
	Consumer(name string) (NatsConsumer, error)
	MustConsumer(name string) NatsConsumer
	ConsumerNames() []string
	GetJetStream() (jetstream.JetStream, error)
	GetConn() (*nats.Conn, error)
	Close()
}

type NatsAlter 

type NatsAlter struct {
	Description string
	PoolCode    string
	Kind        AlterKind
	Safety      AlterSafety
	// contains filtered or unexported fields
}

func GetNatsAlters 

func GetNatsAlters(ctx Context) ([]NatsAlter, error)

GetNatsAlters compares registered NATS stream/consumer definitions with the actual broker state and returns the operations needed to synchronize them. Manages the ORM_ENTITY and ORM_TASK streams plus streams registered with RegisterNatsStream. Only ORM_-prefixed streams are ever deleted.

func (NatsAlter) Exec 

func (a NatsAlter) Exec(ctx Context) error

func (NatsAlter) IsSafe 

func (a NatsAlter) IsSafe() bool

type NatsAuthConfig 

type NatsAuthConfig struct {
	Token     string
	User      string
	Password  string
	CredsFile string
	NKeySeed  string
}

type NatsBatch 

type NatsBatch struct {
	// contains filtered or unexported fields
}

func (NatsBatch) EachError 

func (b NatsBatch) EachError(fn func(error))

func (NatsBatch) EachRecord 

func (b NatsBatch) EachRecord(fn func(*NatsMessage))

func (NatsBatch) Error 

func (b NatsBatch) Error() error

func (NatsBatch) IsEmpty 

func (b NatsBatch) IsEmpty() bool

func (NatsBatch) Records 

func (b NatsBatch) Records() []*NatsMessage

type NatsConsumer 

type NatsConsumer interface {
	GetName() string
	GetSettings() *NatsConsumerSettings
	Fetch(ctx Context, batch int, maxWait time.Duration) NatsBatch
	Close()
}

type NatsConsumerBuilder 

type NatsConsumerBuilder struct {
	// contains filtered or unexported fields
}

func NewNatsConsumer 

func NewNatsConsumer(name, poolCode string) *NatsConsumerBuilder

func (*NatsConsumerBuilder) AckWait 

func (b *NatsConsumerBuilder) AckWait(d time.Duration) *NatsConsumerBuilder

func (*NatsConsumerBuilder) DeliverPolicy 

func (b *NatsConsumerBuilder) DeliverPolicy(p jetstream.DeliverPolicy) *NatsConsumerBuilder

func (*NatsConsumerBuilder) FilterSubjects 

func (b *NatsConsumerBuilder) FilterSubjects(subjects ...string) *NatsConsumerBuilder

func (*NatsConsumerBuilder) MaxAckPending 

func (b *NatsConsumerBuilder) MaxAckPending(n int) *NatsConsumerBuilder

func (*NatsConsumerBuilder) MaxDeliver 

func (b *NatsConsumerBuilder) MaxDeliver(n int) *NatsConsumerBuilder

func (*NatsConsumerBuilder) Stream 

func (b *NatsConsumerBuilder) Stream(name string) *NatsConsumerBuilder

Stream pins the consumer to a stream by name. Without it the reconciler has to guess from the first filter subject, which stops working the moment a consumer filters more than one.

type NatsConsumerSettings 

type NatsConsumerSettings struct {
	Name           string
	FilterSubjects []string
	AckWait        time.Duration
	MaxAckPending  int
	MaxDeliver     int
	DeliverPolicy  jetstream.DeliverPolicy
}

type NatsMessage 

type NatsMessage struct {
	Subject   string
	Data      []byte
	Headers   nats.Header
	Sequence  uint64
	Timestamp time.Time

	// Deliveries is how many times JetStream has delivered this message,
	// starting at 1. A consumer counts retry attempts with it; it is zero on a
	// message you built for publishing.
	Deliveries uint64
	// contains filtered or unexported fields
}

func NewNatsMessage 

func NewNatsMessage(subject string) *NatsMessage

func (*NatsMessage) Ack 

func (m *NatsMessage) Ack() error

func (*NatsMessage) Nak 

func (m *NatsMessage) Nak() error

func (*NatsMessage) NakWithDelay 

func (m *NatsMessage) NakWithDelay(d time.Duration) error

NakWithDelay asks for redelivery after d. Plain Nak redelivers immediately, which turns a permanently failing handler into a hot loop, so a backoff ladder has to use this instead.

func (*NatsMessage) Term 

func (m *NatsMessage) Term() error

type NatsPoolOptions 

type NatsPoolOptions struct {
	ClientID             string
	MaxReconnects        int
	ReconnectWait        time.Duration
	ReconnectBufSize     int
	PingInterval         time.Duration
	ConnectTimeout       time.Duration
	RetryOnFailedConnect bool
	Auth                 *NatsAuthConfig
	IgnoredSubjects      []string
	IgnoredConsumers     []string
}

type NatsPubAck 

type NatsPubAck struct {
	Stream    string
	Sequence  uint64
	Duplicate bool
}

NatsPubAck is the server's answer to a publish. Duplicate reports that the message carried a Nats-Msg-Id the stream had already seen inside its DuplicateWindow, so it was collapsed and will never be delivered - a caller that recorded the publish before making it needs to know.

type NatsStreamBuilder 

type NatsStreamBuilder struct {
	// contains filtered or unexported fields
}

func NewNatsStream 

func NewNatsStream(streamName, poolCode string) *NatsStreamBuilder

func (*NatsStreamBuilder) Duplicates 

func (b *NatsStreamBuilder) Duplicates(d time.Duration) *NatsStreamBuilder

func (*NatsStreamBuilder) MaxAge 

func (b *NatsStreamBuilder) MaxAge(d time.Duration) *NatsStreamBuilder

func (*NatsStreamBuilder) MaxBytes 

func (b *NatsStreamBuilder) MaxBytes(n int64) *NatsStreamBuilder

func (*NatsStreamBuilder) MaxMsgSize 

func (b *NatsStreamBuilder) MaxMsgSize(n int32) *NatsStreamBuilder

func (*NatsStreamBuilder) Replicas 

func (b *NatsStreamBuilder) Replicas(n int) *NatsStreamBuilder

func (*NatsStreamBuilder) Retention 

func (b *NatsStreamBuilder) Retention(r jetstream.RetentionPolicy) *NatsStreamBuilder

func (*NatsStreamBuilder) Storage 

func (b *NatsStreamBuilder) Storage(s jetstream.StorageType) *NatsStreamBuilder

func (*NatsStreamBuilder) Subjects 

func (b *NatsStreamBuilder) Subjects(subjects ...string) *NatsStreamBuilder

type NotFoundError 

type NotFoundError struct {
	Entity string
	ID     uint64
}

NotFoundError is returned by GetByID, SearchOne and reference getters when no row matches. ID is 0 for search and unset-reference misses.

func (*NotFoundError) Error 

func (e *NotFoundError) Error() string

type NullableBoolField 

type NullableBoolField struct{ Column string }

NullableBoolField represents a nullable boolean column.

func (NullableBoolField) ColumnName 

func (f NullableBoolField) ColumnName() string

func (NullableBoolField) Is 

func (f NullableBoolField) Is(v bool) Condition

func (NullableBoolField) IsNotNull 

func (f NullableBoolField) IsNotNull() Condition

func (NullableBoolField) IsNull 

func (f NullableBoolField) IsNull() Condition

type NullableEnumField 

type NullableEnumField struct{ Column string }

NullableEnumField represents a nullable enum column stored as a string.

func (NullableEnumField) ColumnName 

func (f NullableEnumField) ColumnName() string

func (NullableEnumField) In 

func (f NullableEnumField) In(values ...any) Condition

func (NullableEnumField) Is 

func (f NullableEnumField) Is(v any) Condition

func (NullableEnumField) IsNotNull 

func (f NullableEnumField) IsNotNull() Condition

func (NullableEnumField) IsNull 

func (f NullableEnumField) IsNull() Condition

type NullableFloatField 

type NullableFloatField struct{ Column string }

NullableFloatField represents a nullable floating-point column.

func (NullableFloatField) ColumnName 

func (f NullableFloatField) ColumnName() string

func (NullableFloatField) Eq 

func (f NullableFloatField) Eq(v float64) Condition

func (NullableFloatField) Gt 

func (f NullableFloatField) Gt(v float64) Condition

func (NullableFloatField) Gte 

func (f NullableFloatField) Gte(v float64) Condition

func (NullableFloatField) In 

func (f NullableFloatField) In(values ...float64) Condition

func (NullableFloatField) IsNotNull 

func (f NullableFloatField) IsNotNull() Condition

func (NullableFloatField) IsNull 

func (f NullableFloatField) IsNull() Condition

func (NullableFloatField) Lt 

func (f NullableFloatField) Lt(v float64) Condition

func (NullableFloatField) Lte 

func (f NullableFloatField) Lte(v float64) Condition

type NullableIntField 

type NullableIntField struct{ Column string }

NullableIntField represents a nullable signed integer column.

func (NullableIntField) ColumnName 

func (f NullableIntField) ColumnName() string

func (NullableIntField) Eq 

func (f NullableIntField) Eq(v int64) Condition

func (NullableIntField) Gt 

func (f NullableIntField) Gt(v int64) Condition

func (NullableIntField) Gte 

func (f NullableIntField) Gte(v int64) Condition

func (NullableIntField) In 

func (f NullableIntField) In(values ...int64) Condition

func (NullableIntField) IsNotNull 

func (f NullableIntField) IsNotNull() Condition

func (NullableIntField) IsNull 

func (f NullableIntField) IsNull() Condition

func (NullableIntField) Lt 

func (f NullableIntField) Lt(v int64) Condition

func (NullableIntField) Lte 

func (f NullableIntField) Lte(v int64) Condition

type NullableReferenceField 

type NullableReferenceField struct{ Column string }

NullableReferenceField represents a nullable foreign key reference column.

func (NullableReferenceField) ColumnName 

func (f NullableReferenceField) ColumnName() string

func (NullableReferenceField) Eq 

func (f NullableReferenceField) Eq(v uint64) Condition

func (NullableReferenceField) In 

func (f NullableReferenceField) In(values ...uint64) Condition

func (NullableReferenceField) IsNotNull 

func (f NullableReferenceField) IsNotNull() Condition

func (NullableReferenceField) IsNull 

func (f NullableReferenceField) IsNull() Condition

type NullableStringField 

type NullableStringField struct{ Column string }

NullableStringField represents a nullable string column.

func (NullableStringField) ColumnName 

func (f NullableStringField) ColumnName() string

func (NullableStringField) In 

func (f NullableStringField) In(values ...string) Condition

func (NullableStringField) Is 

func (f NullableStringField) Is(v string) Condition

func (NullableStringField) IsEmpty 

func (f NullableStringField) IsEmpty() Condition

func (NullableStringField) IsNotNull 

func (f NullableStringField) IsNotNull() Condition

func (NullableStringField) IsNull 

func (f NullableStringField) IsNull() Condition

func (NullableStringField) Like 

func (f NullableStringField) Like(v string) Condition

type NullableTimeField 

type NullableTimeField struct{ Column string }

NullableTimeField represents a nullable time.Time column.

func (NullableTimeField) ColumnName 

func (f NullableTimeField) ColumnName() string

func (NullableTimeField) Eq 

func (f NullableTimeField) Eq(v time.Time) Condition

func (NullableTimeField) Gt 

func (f NullableTimeField) Gt(v time.Time) Condition

func (NullableTimeField) Gte 

func (f NullableTimeField) Gte(v time.Time) Condition

func (NullableTimeField) IsNotNull 

func (f NullableTimeField) IsNotNull() Condition

func (NullableTimeField) IsNull 

func (f NullableTimeField) IsNull() Condition

func (NullableTimeField) Lt 

func (f NullableTimeField) Lt(v time.Time) Condition

func (NullableTimeField) Lte 

func (f NullableTimeField) Lte(v time.Time) Condition

type NullableUintField 

type NullableUintField struct{ Column string }

NullableUintField represents a nullable unsigned integer column.

func (NullableUintField) ColumnName 

func (f NullableUintField) ColumnName() string

func (NullableUintField) Eq 

func (f NullableUintField) Eq(v uint64) Condition

func (NullableUintField) Gt 

func (f NullableUintField) Gt(v uint64) Condition

func (NullableUintField) Gte 

func (f NullableUintField) Gte(v uint64) Condition

func (NullableUintField) In 

func (f NullableUintField) In(values ...uint64) Condition

func (NullableUintField) IsNotNull 

func (f NullableUintField) IsNotNull() Condition

func (NullableUintField) IsNull 

func (f NullableUintField) IsNull() Condition

func (NullableUintField) Lt 

func (f NullableUintField) Lt(v uint64) Condition

func (NullableUintField) Lte 

func (f NullableUintField) Lte(v uint64) Condition

type Pager 

type Pager struct {
	CurrentPage int
	PageSize    int
}

func NewPager 

func NewPager(currentPage, pageSize int) *Pager

func (*Pager) GetCurrentPage 

func (pager *Pager) GetCurrentPage() int

func (*Pager) GetPageSize 

func (pager *Pager) GetPageSize() int

func (*Pager) IncrementPage 

func (pager *Pager) IncrementPage()

func (*Pager) String 

func (pager *Pager) String() string

type PipeLineBool 

type PipeLineBool struct {
	// contains filtered or unexported fields
}

func (*PipeLineBool) Result 

func (c *PipeLineBool) Result() (bool, error)

type PipeLineGet 

type PipeLineGet struct {
	// contains filtered or unexported fields
}

func (*PipeLineGet) Result 

func (c *PipeLineGet) Result() (value string, has bool, err error)

type PipeLineInt 

type PipeLineInt struct {
	// contains filtered or unexported fields
}

func (*PipeLineInt) Result 

func (c *PipeLineInt) Result() (int64, error)

type PipeLineSlice 

type PipeLineSlice struct {
	// contains filtered or unexported fields
}

func (*PipeLineSlice) Result 

func (c *PipeLineSlice) Result() ([]string, error)

type PipeLineString 

type PipeLineString struct {
	// contains filtered or unexported fields
}

func (*PipeLineString) Result 

func (c *PipeLineString) Result() (string, error)

type PostCommitError 

type PostCommitError struct {
	Err error
}

PostCommitError wraps a failure that happened after the SQL transaction was already committed - cache invalidation, search indexing, CDC publishing or an after-* handler. The rows are durable; retrying the whole operation would apply them twice.

func (*PostCommitError) Error 

func (e *PostCommitError) Error() string

func (*PostCommitError) Unwrap 

func (e *PostCommitError) Unwrap() error

type QueryLoggerOptions 

type QueryLoggerOptions struct {
	MySQL      bool
	Redis      bool
	Clickhouse bool
	Nats       bool
}

type QueryLoggerSource 

type QueryLoggerSource int

type Queue 

type Queue string

Queue routes tasks to the consumer that drains them. It is a routing label and nothing else: the transport tuning belongs to the consumer that declares the queue, and the retention to the one task stream.

type Queued 

type Queued interface {
	Queue() Queue
}

Queued is the optional interface a task implements to pick its queue. Without it the task lands on DefaultQueue. 

func (DownloadSocialAvatar) Queue() orm.Queue { return topology.QueueMedia }

type RedisCache 

type RedisCache interface {
	Set(ctx Context, key string, value any, expiration time.Duration) error
	MSet(ctx Context, pairs ...any) error
	Del(ctx Context, keys ...string) error
	HSet(ctx Context, key string, values ...any) error
	HDel(ctx Context, key string, keys ...string) error
	GetSet(ctx Context, key string, expiration time.Duration, provider func() any) (any, error)
	Info(ctx Context, section ...string) (string, error)
	GetConfig() RedisPoolConfig
	Get(ctx Context, key string) (value string, has bool, err error)
	Eval(ctx Context, script string, keys []string, args ...any) (any, error)
	EvalSha(ctx Context, sha1 string, keys []string, args ...any) (res any, exists bool, err error)
	SetNX(ctx Context, key string, value any, expiration time.Duration) (bool, error)
	ScriptExists(ctx Context, sha1 string) (bool, error)
	ScriptLoad(ctx Context, script string) (string, error)
	LPush(ctx Context, key string, values ...any) (int64, error)
	LPop(ctx Context, key string) (string, error)
	RPush(ctx Context, key string, values ...any) (int64, error)
	LLen(ctx Context, key string) (int64, error)
	Exists(ctx Context, keys ...string) (int64, error)
	Type(ctx Context, key string) (string, error)
	LRange(ctx Context, key string, start, stop int64) ([]string, error)
	LIndex(ctx Context, key string, index int64) (string, bool, error)
	LSet(ctx Context, key string, index int64, value any) error
	RPop(ctx Context, key string) (value string, found bool, err error)
	BLMove(ctx Context, source, destination, srcPos, destPos string, timeout time.Duration) (string, error)
	LMove(ctx Context, source, destination, srcPos, destPos string) (string, error)
	LRem(ctx Context, key string, count int64, value any) error
	Ltrim(ctx Context, key string, start, stop int64) error
	HSetNx(ctx Context, key, field string, value any) (bool, error)
	HMGet(ctx Context, key string, fields ...string) (map[string]any, error)
	HGetAll(ctx Context, key string) (map[string]string, error)
	HGet(ctx Context, key, field string) (value string, has bool, err error)
	HLen(ctx Context, key string) (int64, error)
	HIncrBy(ctx Context, key, field string, incr int64) (int64, error)
	IncrBy(ctx Context, key string, incr int64) (int64, error)
	Incr(ctx Context, key string) (int64, error)
	IncrWithExpire(ctx Context, key string, expire time.Duration) (int64, error)
	Expire(ctx Context, key string, expiration time.Duration) (bool, error)
	ZAdd(ctx Context, key string, members ...redis.Z) (int64, error)
	ZRevRange(ctx Context, key string, start, stop int64) ([]string, error)
	ZRevRangeWithScores(ctx Context, key string, start, stop int64) ([]redis.Z, error)
	ZRangeWithScores(ctx Context, key string, start, stop int64) ([]redis.Z, error)
	ZRangeArgsWithScores(ctx Context, z redis.ZRangeArgs) ([]redis.Z, error)
	ZCard(ctx Context, key string) (int64, error)
	ZCount(ctx Context, key string, min, max string) (int64, error)
	ZScore(ctx Context, key, member string) (float64, error)
	MGet(ctx Context, keys ...string) ([]any, error)
	SAdd(ctx Context, key string, members ...any) (int64, error)
	SMembers(ctx Context, key string) ([]string, error)
	SIsMember(ctx Context, key string, member any) (bool, error)
	SCard(ctx Context, key string) (int64, error)
	SPop(ctx Context, key string) (string, bool, error)
	SPopN(ctx Context, key string, max int64) ([]string, error)
	XTrim(ctx Context, stream string, maxLen int64) (deleted int64, err error)
	XRange(ctx Context, stream, start, stop string, count int64) ([]redis.XMessage, error)
	XRevRange(ctx Context, stream, start, stop string, count int64) ([]redis.XMessage, error)
	XInfoStream(ctx Context, stream string) (*redis.XInfoStream, error)
	XInfoGroups(ctx Context, stream string) ([]redis.XInfoGroup, error)
	XGroupCreate(ctx Context, stream, group, start string) (key string, exists bool, err error)
	XGroupCreateMkStream(ctx Context, stream, group, start string) (key string, exists bool, err error)
	XGroupDestroy(ctx Context, stream, group string) (int64, error)
	XRead(ctx Context, a *redis.XReadArgs) ([]redis.XStream, error)
	XDel(ctx Context, stream string, ids ...string) (int64, error)
	XAutoClaim(ctx Context, a *redis.XAutoClaimArgs) (messages []redis.XMessage, start string, err error)
	XGroupDelConsumer(ctx Context, stream, group, consumer string) (int64, error)
	XReadGroup(ctx Context, a *redis.XReadGroupArgs) (streams []redis.XStream, err error)
	XPending(ctx Context, stream, group string) (*redis.XPending, error)
	XPendingExt(ctx Context, a *redis.XPendingExtArgs) ([]redis.XPendingExt, error)
	XLen(ctx Context, stream string) (int64, error)
	XClaim(ctx Context, a *redis.XClaimArgs) ([]redis.XMessage, error)
	XClaimJustID(ctx Context, a *redis.XClaimArgs) ([]string, error)
	XAck(ctx Context, stream, group string, ids ...string) (int64, error)
	XAckDel(ctx Context, stream, group string, ids ...string) ([]interface{}, error)
	FlushAll(ctx Context) error
	FlushDB(ctx Context) error
	Scan(ctx Context, cursor uint64, match string, count int64) (keys []string, cursorNext uint64, err error)
	GetLocker() *Locker
	GetCode() string
	Client() *redis.Client
}

type RedisCacheEntityProvider 

type RedisCacheEntityProvider interface {
	RedisCode() string
	RedisCachePrefix() string
	ClearRedisCache(ctx Context) (int, error)
}

RedisCacheEntityProvider is implemented by providers whose entity has a Redis cache (redisCache tag).

type RedisOptions 

type RedisOptions struct {
	User            string
	Password        string
	Master          string
	Sentinels       []string
	SentinelOptions *redis.FailoverOptions
}

type RedisPipeLine 

type RedisPipeLine struct {
	// contains filtered or unexported fields
}

func (*RedisPipeLine) Del 

func (rp *RedisPipeLine) Del(key ...string)

func (*RedisPipeLine) Exec 

func (rp *RedisPipeLine) Exec(ctx Context) (response []redis.Cmder, err error)

func (*RedisPipeLine) Expire 

func (rp *RedisPipeLine) Expire(key string, expiration time.Duration) *PipeLineBool

func (*RedisPipeLine) Get 

func (rp *RedisPipeLine) Get(key string) *PipeLineGet

func (*RedisPipeLine) HDel 

func (rp *RedisPipeLine) HDel(key string, values ...string)

func (*RedisPipeLine) HIncrBy 

func (rp *RedisPipeLine) HIncrBy(key, field string, incr int64) *PipeLineInt

func (*RedisPipeLine) HSet 

func (rp *RedisPipeLine) HSet(key string, values ...any)

func (*RedisPipeLine) LPush 

func (rp *RedisPipeLine) LPush(key string, values ...any)

func (*RedisPipeLine) LRange 

func (rp *RedisPipeLine) LRange(key string, start, stop int64) *PipeLineSlice

func (*RedisPipeLine) LRangeBatchInto 

func (rp *RedisPipeLine) LRangeBatchInto(results []PipeLineSlice, keys []string, start, stop int64)

LRangeBatchInto queues one LRANGE per key and fills caller-owned result handles. The result and key slices must have equal lengths. Commands keep independent results after Exec; their temporary command storage is shared within the batch.

func (*RedisPipeLine) LRangeValue 

func (rp *RedisPipeLine) LRangeValue(key string, start, stop int64) PipeLineSlice

LRangeValue queues LRANGE and returns its result handle by value, allowing generated batch readers to store handles without allocating a wrapper per key.

func (*RedisPipeLine) MSet 

func (rp *RedisPipeLine) MSet(pairs ...any)

func (*RedisPipeLine) RPush 

func (rp *RedisPipeLine) RPush(key string, values ...any)

func (*RedisPipeLine) SAdd 

func (rp *RedisPipeLine) SAdd(key string, members ...any)

func (*RedisPipeLine) SRem 

func (rp *RedisPipeLine) SRem(key string, members ...any)

func (*RedisPipeLine) Set 

func (rp *RedisPipeLine) Set(key string, value any, expiration time.Duration)

type RedisPoolConfig 

type RedisPoolConfig interface {
	GetCode() string
	GetDatabaseNumber() int
	GetAddress() string
	// contains filtered or unexported methods
}

type Reference 

type Reference[E any] uint64

type ReferenceField 

type ReferenceField struct{ Column string }

ReferenceField represents a foreign key reference column.

func (ReferenceField) ColumnName 

func (f ReferenceField) ColumnName() string

func (ReferenceField) Eq 

func (f ReferenceField) Eq(v uint64) Condition

func (ReferenceField) In 

func (f ReferenceField) In(values ...uint64) Condition

type References 

type References[E any] struct {
	// contains filtered or unexported fields
}

type Registry 

type Registry interface {
	Validate() (Engine, error)
	ValidateForCodeGen() (Engine, error)
	RegisterEntity(entity ...any)
	RegisterMySQL(dataSourceName string, poolCode string, poolOptions *MySQLOptions)
	RegisterRedis(address string, db int, poolCode string, options *RedisOptions)
	InitByYaml(yaml any) error
	InitByConfig(config *Config) error
	SetOption(key string, value any)
	RegisterClickhouse(dataSourceName string, poolCode string, poolOptions *ClickhouseOptions)
	RegisterClickhouseTable(table *ClickhouseTableBuilder)
	RegisterNats(urls []string, poolCode string, options *NatsPoolOptions)
	RegisterNatsStream(stream *NatsStreamBuilder)
	RegisterNatsConsumer(consumer *NatsConsumerBuilder)
	RegisterConsumer(def ConsumerDef)
	RegisterEntityStream(opts EntityStreamOptions)
	RegisterTask(task any, opts TaskOptions)
	RegisterTaskStream(opts TaskStreamOptions)
	EnableMetrics(factory promauto.Factory)
}

func NewRegistry 

func NewRegistry() Registry

type Rows 

type Rows interface {
	Next() bool
	Scan(dest ...any) error
	Columns() ([]string, error)
}

type SQLRow 

type SQLRow interface {
	Scan(dest ...any) error
	Err() error
}

type SQLRows 

type SQLRows interface {
	Next() bool
	Err() error
	Close() error
	Scan(dest ...any) error
	Columns() ([]string, error)
}

type StreamConsumer 

type StreamConsumer interface {
	// Consume fetches up to `batch` messages with `timeout` deadline, dispatches
	// each one, acks on nil error and leaves unacked on err. It returns the first
	// transport-level error (connection lost, stream gone, durable missing).
	// Handler errors are reported to orm's NATS log but do not return from
	// Consume - they only block the ack so JetStream can redeliver.
	Consume(ctx context.Context, batch int, timeout time.Duration) error
}

StreamConsumer is the low-level pull primitive. Callers loop over Consume() in whatever lifecycle harness they use (cron.Job, fx lifecycle, plain goroutine). orm intentionally does NOT provide a blocking Run() - the application owns the loop, the retry policy and the shutdown semantics.

type StringField 

type StringField struct{ Column string }

StringField represents a string column.

func (StringField) ColumnName 

func (f StringField) ColumnName() string

func (StringField) In 

func (f StringField) In(values ...string) Condition

func (StringField) Is 

func (f StringField) Is(v string) Condition

func (StringField) IsEmpty 

func (f StringField) IsEmpty() Condition

func (StringField) Like 

func (f StringField) Like(v string) Condition

type Subject 

type Subject string

Subject is the identity of something that happened. Entity changes get one subject per table; a consumer selects the subset it wants with a subject filter. That is what makes one write one message however many consumers read it - fan-out happens at subscribe time, not by duplicating bytes at publish time.

func EntitySubject 

func EntitySubject(table string) Subject

EntitySubject is where an entity's changes are published. The token is the table name, not the Go type name, so renaming the entity struct does not move the subject and orphan a consumer's filter.

func ReplaySubject 

func ReplaySubject(consumer ConsumerName, table string) Subject

ReplaySubject addresses a synthetic event to one consumer only.

This is the reason replay gets its own subject space rather than reusing the entity subject: republishing an order onto orm.entity.orders would reach every consumer of orders, so reindexing Elasticsearch would re-fire the customer notifications that another consumer sends. A private subject makes replay per-consumer by construction.

func TaskSubject 

func TaskSubject(queue Queue, task TaskName) Subject

TaskSubject is where one task's dispatches live. Task names are Go type names and NATS subject tokens are case-sensitive, so no transform is needed.

func (Subject) String 

func (s Subject) String() string

type Subjected 

type Subjected interface {
	Subject() Subject
}

Subjected is the optional full override of a task's wire subject. Its one real use is pinning the old subject across a Go rename, because messages already on the stream carry the old name.

The subject must stay under `orm.task.`; Validate() rejects anything else.

type TXClient 

type TXClient interface {
	DBClientQuery
}

type TaskName 

type TaskName string

TaskName is a task's wire name, which is its Go type name.

func TaskNameFor 

func TaskNameFor[T any]() TaskName

TaskNameFor is a task's wire name derived from its type, so a caller with a generic task parameter never has to repeat it as a string. It matches what newTaskSchema registers, which is what makes the two impossible to disagree.

type TaskOptions 

type TaskOptions struct {
	MaxAttempts int
	BaseBackoff time.Duration
	MaxBackoff  time.Duration
}

TaskOptions tunes one task's delivery. Every field has a default, so RegisterTask(X{}, TaskOptions{}) is a valid registration.

Identity is deliberately absent: the queue comes from Queued and the subject from the type name or Subjected, so no two places can disagree about what a task is called.

type TaskStreamOptions 

type TaskStreamOptions struct {
	Storage         jetstream.StorageType
	Replicas        int
	MaxAge          time.Duration
	MaxBytes        int64
	DuplicateWindow time.Duration
	NatsPool        string
}

TaskStreamOptions tunes the single stream that carries dispatched tasks.

type TimeField 

type TimeField struct{ Column string }

TimeField represents a time.Time column.

func (TimeField) ColumnName 

func (f TimeField) ColumnName() string

func (TimeField) Eq 

func (f TimeField) Eq(v time.Time) Condition

func (TimeField) Gt 

func (f TimeField) Gt(v time.Time) Condition

func (TimeField) Gte 

func (f TimeField) Gte(v time.Time) Condition

func (TimeField) Lt 

func (f TimeField) Lt(v time.Time) Condition

func (TimeField) Lte 

func (f TimeField) Lte(v time.Time) Condition

type UintField 

type UintField struct{ Column string }

UintField represents an unsigned integer column.

func (UintField) ColumnName 

func (f UintField) ColumnName() string

func (UintField) Eq 

func (f UintField) Eq(v uint64) Condition

func (UintField) Gt 

func (f UintField) Gt(v uint64) Condition

func (UintField) Gte 

func (f UintField) Gte(v uint64) Condition

func (UintField) In 

func (f UintField) In(values ...uint64) Condition

func (UintField) Lt 

func (f UintField) Lt(v uint64) Condition

func (UintField) Lte 

func (f UintField) Lte(v uint64) Condition

type Where 

type Where interface {
	String() string
	GetParameters() []any
	IsWithFakeDeletes() bool
}

Source Files

View all Source files

Directories

Path Synopsis
test_fixtures
jobtasks
Package jobtasks holds task fixtures.
 Package jobtasks holds task fixtures.
media/jobtasks
Package jobtasks deliberately shares its name with the other fixture package so the generator's import aliasing has something to disambiguate.
 Package jobtasks deliberately shares its name with the other fixture package so the generator's import aliasing has something to disambiguate.
genboot command
Command genboot regenerates test_generate/entities without compiling the test binary.
 Command genboot regenerates test_generate/entities without compiling the test binary.
models

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.