db

package
v0.55.1 Latest Latest
Warning

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

 Go to latest Published: Jul 11, 2026 License: Apache-2.0 Imports: 17 Imported by: 0

Documentation

Overview

Package db owns the postgres connection pool, migrations, and per-table repositories.

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrDeviceCodePending = errors.New("device_code not yet verified")
	ErrDeviceCodeExpired = errors.New("device_code expired")
)
View Source 
var ErrAlreadyMember = errors.New("user is already a member of this account")

ErrAlreadyMember is returned by CreateInvite when the invitee is already a member or the owner of the org. Same intent as the duplicate-pending case but a different cause — surface separately.

View Source 
var ErrCannotRemoveOwner = errors.New("cannot remove the account's owner")

ErrCannotRemoveOwner is returned by RemoveMember when the caller tries to remove the org's owner. v1 has no transfer-ownership path; the owner has to stay on the org until v2 lands transfer.

View Source 
var ErrDuplicatePendingInvite = errors.New("a pending invite for this user already exists")

ErrDuplicatePendingInvite is returned by CreateInvite when an active pending invite for the same (org, username) already exists. The unique partial index enforces this server-side; we surface a typed sentinel so handlers can return a clean 409.

View Source 
var ErrHandleTaken = errors.New("handle taken")

ErrHandleTaken is returned when a (org, handle) row already exists.

View Source 
var ErrInvalidUserMode = errors.New("user mode must be 'github' or 'internal'")

ErrInvalidUserMode is returned when a caller passes a Mode value outside the {github, internal} CHECK constraint.

View Source 
var ErrInviteNotPending = errors.New("invite is not pending")

ErrInviteNotPending is returned when a transition (accept/decline/ revoke) targets an invite that is no longer pending. Caller surfaces it as 409.

View Source 
var ErrNotFound = errors.New("not found")

ErrNotFound is returned by lookups when the row does not exist.

View Source 
var ErrPipeNameTaken = errors.New("pipe name taken")

ErrPipeNameTaken — uniqueness collision on insert. The partial UNIQUE indexes split this by collared/uncollared shape, but both surface here.

View Source 
var ErrScheduleIDAmbiguous = errors.New("schedule short id is ambiguous (matches multiple schedules)")

ErrScheduleIDAmbiguous — a short id matched more than one row in the account. InsertSchedule regenerates uuids to avoid ever minting a colliding suffix, so this is a should-never belt; `schedule rm` must still refuse to guess rather than silently delete multiple rows.

Functions

func AcceptInvite added in v0.22.0 

func AcceptInvite(ctx context.Context, p *Pool, inviteID, acceptingUserID uuid.UUID) error

AcceptInvite transitions a pending invite to accepted and adds the accepting user as a non-owner member of the org, atomically. The caller passes the accepting user's id; the function checks that their username matches the invite's invitee_username (so a logged-in user can't accept someone else's invite even if they know the id).

Returns:

  • ErrNotFound if the invite id doesn't exist
  • ErrInviteNotPending if the invite is no longer pending
  • errors.New("forbidden") if userID's username != invitee_username

func AddMember 

func AddMember(ctx context.Context, p *Pool, accountID, userID uuid.UUID) error

AddMember records a user as a non-owner member of an org. Idempotent — re-adding an existing member is a no-op (UPSERT-style).

func ApproveDeviceCode 

func ApproveDeviceCode(ctx context.Context, p *Pool, userCode string, userID, accountID uuid.UUID) error

ApproveDeviceCode marks the user_code as verified by userID, binding it to the org accountID the user selected on the verify page. Idempotent. Returns ErrNotFound if user_code doesn't exist or is already expired.

func BumpScheduleFailCount added in v0.51.0 

func BumpScheduleFailCount(ctx context.Context, p *Pool, id uuid.UUID) (int, error)

BumpScheduleFailCount increments the consecutive-failure counter after a failed publish and returns the post-bump count — the value the scheduler compares against its drop threshold.

func ChatCursorKey added in v0.55.0 

func ChatCursorKey(kind, acting, target string) string

ChatCursorKey composes the (kind, acting, target) conversation key the read-cursor table is indexed by, so callers build map keys the same way the roster lookup does. `acting` is the viewer's handle ("" for pipes / god's-eye).

func CompleteFire added in v0.51.0 

func CompleteFire(ctx context.Context, p *Pool, id uuid.UUID, next *time.Time, firedAt time.Time) error

CompleteFire settles a claimed row: next==nil deletes it (spent one-off), otherwise next_fire_at is set to *next. A non-zero firedAt stamps last_fired_at and resets fail_count (the counter tracks CONSECUTIVE failures); a beyond-grace skip advances without touching either.

func ConsumeDeviceCode 

func ConsumeDeviceCode(ctx context.Context, p *Pool, deviceCode string) (uuid.UUID, *uuid.UUID, error)

ConsumeDeviceCode is the CLI poll path. Returns the user_id and the org the user selected (nil if approved without one) and marks the code consumed (single-use). State machine errors:

  • ErrDeviceCodePending if not yet verified
  • ErrDeviceCodeExpired if past TTL
  • ErrNotFound if unknown / already consumed

func DeclineInvite added in v0.22.0 

func DeclineInvite(ctx context.Context, p *Pool, inviteID, decliningUserID uuid.UUID) error

DeclineInvite transitions a pending invite to declined. Same permission check as Accept: caller's username must match the invite's invitee_username.

func DeleteChatReadCursorsForTarget added in v0.55.0 

func DeleteChatReadCursorsForTarget(ctx context.Context, p *Pool, accountID uuid.UUID, kind, target string) error

DeleteChatReadCursorsForTarget removes every read cursor (all users, all acting handles) for one window. Called when the underlying pipe/source is deleted so a same-name recreate starts fresh instead of inheriting a stale high last_read_seq (which would suppress the new stream's early unread).

func DeletePipe 

func DeletePipe(ctx context.Context, p *Pool, sourceID uuid.UUID, name string) error

DeletePipe removes the row. Returns ErrNotFound when (source, name) doesn't exist. Stream cleanup is the caller's responsibility (server-side).

func DeleteScheduleByShortID added in v0.51.0 

func DeleteScheduleByShortID(ctx context.Context, p *Pool, accountID uuid.UUID, shortID string) (bool, error)

DeleteScheduleByShortID removes one schedule addressed by the CLI-facing short id (last 8 hex of the uuid). Returns whether a row matched — false maps to E_SCHEDULE_NOT_FOUND at the handler. A multi-row match (per-account suffix collision — prevented at insert, guarded here as a belt) deletes nothing and returns ErrScheduleIDAmbiguous. The suffix expression can't use an index; fine at expected per-account row counts.

func DeleteSource added in v0.21.0 

func DeleteSource(ctx context.Context, p *Pool, accountID uuid.UUID, handle string) error

DeleteSource removes a source row. The pipes FK is ON DELETE CASCADE so pipe rows are removed automatically. JetStream stream cleanup is the caller's responsibility. Returns ErrNotFound when (org, handle) doesn't exist.

func DeleteUncollaredPipe added in v0.31.0 

func DeleteUncollaredPipe(ctx context.Context, p *Pool, accountID uuid.UUID, manifold, name string) error

DeleteUncollaredPipe removes an uncollared pipe row by (account, manifold, name). Stream cleanup is the caller's responsibility. Phase 1.5.

func GeneratePlaintextKey 

func GeneratePlaintextKey() (string, error)

GeneratePlaintextKey returns a fresh plaintext API key. Format: "ppz_<26 hex chars>" (a UUIDv7 hex without dashes, prefixed). 30 chars total makes the 8-char display prefix human-meaningful while leaving plenty of entropy.

func GetLastSelectedAccount added in v0.50.1 

func GetLastSelectedAccount(ctx context.Context, p *Pool, userID uuid.UUID) (*uuid.UUID, error)

GetLastSelectedAccount returns the org the user last authorized a CLI session into (nil if they never have, or the org was since deleted — the FK is ON DELETE SET NULL). Used to default the device-flow org dropdown.

func HashAPIKey 

func HashAPIKey(plaintext string) (string, error)

HashAPIKey produces a self-describing argon2id hash: 

$argon2id$v=19$m=65536,t=1,p=4$<base64-salt>$<base64-tag>

func IsMemberOrOwner 

func IsMemberOrOwner(ctx context.Context, p *Pool, accountID, userID uuid.UUID) bool

IsMemberOrOwner returns true if userID owns accountID or is a member. Used by /auth/exchange (Phase 3.5) to validate that a multi-account user is actually entitled to the account they're requesting a JWT for.

func KeyPrefix 

func KeyPrefix(plaintext string) string

KeyPrefix is the first 8 characters AFTER the "ppz_" sentinel. Used for display in the GUI and the `ppz status` line. Never use the prefix for auth.

func ListChatReadCursors added in v0.55.0 

func ListChatReadCursors(ctx context.Context, p *Pool, accountID, userID uuid.UUID) (map[string]int64, error)

ListChatReadCursors returns the user's read position for every conversation in the account as a map keyed by ChatCursorKey(kind, acting, target) -> last_read_seq. Absent conversations (never opened) are missing (seq 0 to the caller). One query feeds the whole roster's unread badges.

func Migrate 

func Migrate(ctx context.Context, p *Pool) error

Migrate runs every migration file in order, lexicographically by name. Each file uses IF NOT EXISTS / IF NOT EXISTS COLUMN clauses so re-running is idempotent.

func PipesExistAtManifold added in v0.31.1 

func PipesExistAtManifold(ctx context.Context, p *Pool, accountID uuid.UUID, manifold string) (bool, error)

PipesExistAtManifold reports whether ANY pipe (collared or uncollared) exists at the given manifold prefix. Phase 1.5.1 collision check — a new source's handle at manifold M reserves the prefix path M.<handle> (or just <handle> if M is empty), so source creation rejects when pipes already live there.

func RemoveMember 

func RemoveMember(ctx context.Context, p *Pool, accountID, userID uuid.UUID) error

RemoveMember drops a non-owner from the org. Returns ErrCannotRemoveOwner when targetUserID matches the org's owner — caller surfaces it as 409. ErrNotFound when the user wasn't a member.

func RevokeAPIKey 

func RevokeAPIKey(ctx context.Context, p *Pool, id uuid.UUID) error

RevokeAPIKey marks the key revoked. Idempotent: revoking an already-revoked key is a no-op (returns nil). Returns ErrNotFound if no row matches the id.

func RevokeInvite added in v0.22.0 

func RevokeInvite(ctx context.Context, p *Pool, inviteID uuid.UUID) error

RevokeInvite transitions a pending invite to revoked. Owner-only: caller is expected to be the org's owner; this function does not re-check the role (handlers gate via owner-only middleware), but it does require the invite be in the pending state.

func SetLastSelectedAccount added in v0.50.1 

func SetLastSelectedAccount(ctx context.Context, p *Pool, userID, accountID uuid.UUID) error

SetLastSelectedAccount records the org a user just authorized into so the next device-flow verify page defaults its dropdown to it.

func SetNATSAccount 

func SetNATSAccount(ctx context.Context, p *Pool, accountID uuid.UUID, accountPub, accountJWT, signingSeed string) error

SetNATSAccount persists the Operator-signed Account JWT + the account's signing seed for an account row. Called once (lazily) on first /auth/exchange after Phase 3.5 — subsequent calls find the row already populated and skip.

func SourceExistsAtManifold added in v0.31.1 

func SourceExistsAtManifold(ctx context.Context, p *Pool, accountID uuid.UUID, manifold, handle string) (bool, error)

SourceExistsAtManifold reports whether a source with the given handle exists at the manifold. Phase 1.5.1 collision check — uncollared pipe creation rejects when a source shares the proposed name at the same manifold.

func UncollaredPipeExists added in v0.31.1 

func UncollaredPipeExists(ctx context.Context, p *Pool, accountID uuid.UUID, manifold, name string) (bool, error)

func UpdateLastBroadcast 

func UpdateLastBroadcast(ctx context.Context, p *Pool, accountID uuid.UUID, handle string, at time.Time, payload string) error

UpdateLastBroadcast records the most recent broadcast for this source. Called by the server-side subscriber on every message. Idempotent on identical inputs.

func UpsertChatReadCursor added in v0.55.0 

func UpsertChatReadCursor(ctx context.Context, p *Pool, accountID, userID uuid.UUID, kind, acting, target string, seq int64) error

UpsertChatReadCursor advances the user's read position for one conversation to seq. GREATEST(existing, excluded) means the cursor only ever moves forward, so a stale/late write can't rewind a read position past what the user has seen.

func UserHasAnyPipe added in v0.31.8 

func UserHasAnyPipe(ctx context.Context, p *Pool, userID uuid.UUID) (bool, error)

UncollaredPipeExists reports whether a sourceless pipe with the given (account, manifold, name) already exists. Phase 1.5.1 collision check — source creation rejects when an uncollared pipe shares the source's proposed name at the same manifold. UserHasAnyPipe reports whether the user owns or is a member of any account that has at least one pipe (collared or uncollared). Powers the dashboard onboarding panel: when this is false, the empty-state get-started instructions render; once the user creates a pipe the panel hides itself.

func UsernamesByIDs added in v0.27.0 

func UsernamesByIDs(ctx context.Context, p *Pool, ids []uuid.UUID) (map[uuid.UUID]string, error)

UsernamesByIDs resolves a set of user IDs to {id → username}. Used by the server's list endpoints that need to attribute every source/pipe row to a user (HUMAN column). Single round-trip via ANY($1::uuid[]). Missing IDs are simply absent from the returned map — callers should treat that as "" so a stale ID can't break rendering.

func VerifyAPIKey 

func VerifyAPIKey(plaintext, stored string) bool

VerifyAPIKey checks plaintext against a stored hash in constant time.

Types

type APIKey 

type APIKey struct {
	ID              uuid.UUID
	AccountID       uuid.UUID
	CreatedByUserID uuid.UUID // user that minted the key (NOT NULL)
	KeyHash         string
	KeyPrefix       string
	Label           string
	CreatedAt       time.Time
	// RevokedAt is nil for active keys, set to the revoke time once
	// `POST /api/v1/keys/<id>/revoke` flips it. LookupAPIKey filters
	// out revoked rows; the GUI shows them with strikethrough so the
	// audit trail stays visible.
	RevokedAt *time.Time
}

func InsertAPIKey 

func InsertAPIKey(ctx context.Context, p *Pool, accountID, createdBy uuid.UUID, label string) (key APIKey, plaintext string, err error)

InsertAPIKey mints a fresh plaintext key, hashes it, and writes the row. `createdBy` is the user who minted the key — required (NOT NULL on the table) so every key is attributable for `ppz ls` HUMAN.

func ListAPIKeysForOrg 

func ListAPIKeysForOrg(ctx context.Context, p *Pool, accountID uuid.UUID) ([]APIKey, error)

ListAPIKeysForOrg returns every key for the org, including revoked ones — the GUI shows revoked keys (with strikethrough) so the audit trail stays visible. Sorted active-first by creation time, then revoked rows.

func LookupAPIKey 

func LookupAPIKey(ctx context.Context, p *Pool, plaintext string) (APIKey, error)

LookupAPIKey resolves a plaintext key to its row by scanning all keys with the matching 8-char prefix and verifying the hash. ErrNotFound when no match (including when the matching key has been revoked).

func (APIKey) Revoked 

func (k APIKey) Revoked() bool

Revoked is a small accessor — useful from html/template, which can't dereference pointers cleanly.

type Account added in v0.30.0 

type Account struct {
	ID          uuid.UUID
	Name        string
	OwnerUserID uuid.UUID
	CreatedAt   time.Time

	// Auth V2 §Phase 3.5 — per-account NATS credential. NULL until the
	// account's NATS credential is provisioned (lazy on first /auth/exchange).
	NATSAccountPub         string
	NATSAccountJWT         string
	NATSAccountSigningSeed string
}

Account is the ppz-side tenancy boundary. Maps 1:1 to a NATS account (see SetNATSAccount below for the credential plumbing). Pre-launch this was called "Organisation"; the rename is part of the Phase 1 surface strip (see strategy doc OSS-PIPESCLOUD-ARCHITECTURE-SPLIT, private, locked decisions #11 and #18).

func DefaultAccountFor added in v0.50.1 

func DefaultAccountFor(ctx context.Context, p *Pool, userID uuid.UUID) (Account, error)

DefaultAccountFor returns the account to use for a caller who hasn't explicitly selected one: their oldest OWNED account, or — if they own none — their oldest account by MEMBERSHIP. Returns ErrNotFound only when the user neither owns nor belongs to any account.

This is the account-selection default for the OAuth path (device flow + /auth/exchange) and the requireAPIKey ?org fallback. It must consider membership: a user invited into an org (a member, not owner) otherwise has no usable default and every OAuth call 403s / errors — and listing (?org=, membership-aware) would disagree with the minted NATS creds.

func GetAccount added in v0.30.0 

func GetAccount(ctx context.Context, p *Pool, id uuid.UUID) (Account, error)

func GetAccountByName added in v0.30.0 

func GetAccountByName(ctx context.Context, p *Pool, name string) (Account, error)

GetAccountByName looks up an account by its unique name (used as a slug alias in the GUI: /accounts/alpha resolves the same as /accounts/<uuid>).

func InsertAccount added in v0.30.0 

func InsertAccount(ctx context.Context, p *Pool, name string, ownerUserID uuid.UUID) (Account, error)

InsertAccount creates a new account owned by ownerUserID. If ownerUserID is uuid.Nil, the account defaults to the seeded unauthenticated user — preserves back-compat for tests + GUI callers that don't supply an owner yet.

func ListAccounts added in v0.30.0 

func ListAccounts(ctx context.Context, p *Pool) ([]Account, error)

func ListAccountsForUser added in v0.30.0 

func ListAccountsForUser(ctx context.Context, p *Pool, userID uuid.UUID) ([]Account, error)

ListAccountsForUser returns the accounts userID owns or is a member of, ordered by name. Used by the GUI dashboard so each user sees only their own tenants instead of every account in the system.

type DeviceCode 

type DeviceCode struct {
	DeviceCode string
	UserCode   string
	ClientName string
	UserID     *uuid.UUID
	AccountID  *uuid.UUID // org the user picked on the verify page
	ExpiresAt  time.Time
	VerifiedAt *time.Time
	ConsumedAt *time.Time
	CreatedAt  time.Time
}

DeviceCode is the row stored in oauth_device_codes.

func CreateDeviceCode 

func CreateDeviceCode(ctx context.Context, p *Pool, ttl time.Duration, clientName string) (DeviceCode, error)

CreateDeviceCode mints a fresh pair, inserts the row, returns it. clientName is a free-form label the CLI sends so the verify page can name the calling app (e.g. "ppz CLI 0.15.0"); empty string is fine — the page falls back to generic copy.

func LookupDeviceCode 

func LookupDeviceCode(ctx context.Context, p *Pool, deviceCode string) (DeviceCode, error)

func LookupDeviceCodeByUserCode 

func LookupDeviceCodeByUserCode(ctx context.Context, p *Pool, userCode string) (DeviceCode, error)

type Invite added in v0.22.0 

type Invite struct {
	ID              uuid.UUID
	AccountID       uuid.UUID
	InviteeUsername string
	InviterUserID   uuid.UUID
	Status          InviteStatus
	CreatedAt       time.Time
	DecidedAt       *time.Time
}

func CreateInvite added in v0.22.0 

func CreateInvite(ctx context.Context, p *Pool, accountID uuid.UUID, inviteeUsername string, inviterUserID uuid.UUID) (Invite, error)

CreateInvite inserts a pending invite for inviteeUsername into accountID, recording inviterUserID as the sender. Pre-flights two error conditions before the insert:

  • inviteeUsername is already a member or the owner of accountID → ErrAlreadyMember
  • a pending invite for the same (org, username) already exists → ErrDuplicatePendingInvite

Pre-flighting the duplicate case in addition to relying on the partial unique index gives us a typed error without inspecting pg error codes.

func GetInvite added in v0.22.0 

func GetInvite(ctx context.Context, p *Pool, id uuid.UUID) (Invite, error)

GetInvite fetches a single invite by id. Returns ErrNotFound if the row doesn't exist.

func ListInvitesForOrg added in v0.22.0 

func ListInvitesForOrg(ctx context.Context, p *Pool, accountID uuid.UUID) ([]Invite, error)

ListInvitesForOrg returns every invite (any status) for accountID, newest first. Used by the org page to show the invite history.

type InviteStatus added in v0.22.0 

type InviteStatus string

InviteStatus values mirror the CHECK constraint on invites.status. 

const (
	InviteStatusPending  InviteStatus = "pending"
	InviteStatusAccepted InviteStatus = "accepted"
	InviteStatusDeclined InviteStatus = "declined"
	InviteStatusRevoked  InviteStatus = "revoked"
)

type InviteWithAccount added in v0.30.0 

type InviteWithAccount struct {
	Invite
	AccountName string
}

InviteWithAccount is the dashboard projection — invite plus org name so the user can see where the invite came from without a second query.

func ListPendingInvitesForUsername added in v0.22.0 

func ListPendingInvitesForUsername(ctx context.Context, p *Pool, username string) ([]InviteWithAccount, error)

ListPendingInvitesForUsername returns all pending invites whose invitee_username matches. Joined with accounts so callers can show the org name on the dashboard.

type Member 

type Member struct {
	AccountID uuid.UUID
	UserID    uuid.UUID
	AddedAt   time.Time
}

type OAuthToken 

type OAuthToken struct {
	ID         uuid.UUID
	UserID     uuid.UUID
	TokenHash  string
	Prefix     string
	ExpiresAt  time.Time
	RevokedAt  *time.Time
	CreatedAt  time.Time
	LastUsedAt *time.Time
}

func IssueBearerToken 

func IssueBearerToken(ctx context.Context, p *Pool, userID uuid.UUID, ttl time.Duration) (string, OAuthToken, error)

func LookupBearerToken 

func LookupBearerToken(ctx context.Context, p *Pool, plaintext string) (OAuthToken, error)

type Pipe 

type Pipe struct {
	ID              uuid.UUID
	AccountID       uuid.UUID  // tenancy anchor (denormalised; matches source.account_id when SourceID is set)
	Manifold        string     // hierarchical-grouping segment; ” = root (NOT NULL on the DB column)
	SourceID        *uuid.UUID // nil for uncollared (sourceless) pipes
	CreatedByUserID uuid.UUID  // user that created the pipe (NOT NULL)
	Name            string
	TTLSeconds      *int   // nil = use server default (86400 s)
	MaxMsgs         *int   // nil = use server default (1000)
	MaxBytes        *int64 // nil = use server default (64 MiB)
	CreatedAt       time.Time
}

Pipe is one user-creatable channel. Phase 1.5: pipes carry an explicit manifold (hierarchical-grouping segment, ” = root) and an account_id (denormalised from source.account_id for collared rows, explicit for uncollared ones). SourceID is nullable — uncollared (sourceless) pipes are symmetric many-to-many channels under a manifold.

Auto-provisioned pipes (stdin, stdout, stdctrl, inbox) are NOT stored here — they're derived from the source's kind and joined in at API response time.

func GetPipeByName 

func GetPipeByName(ctx context.Context, p *Pool, sourceID uuid.UUID, name string) (Pipe, error)

GetPipeByName returns one pipe row or ErrNotFound.

func InsertPipe 

func InsertPipe(ctx context.Context, p *Pool, accountID uuid.UUID, manifold string, sourceID *uuid.UUID, createdBy uuid.UUID, name string, ttl *int, maxMsgs *int, maxBytes *int64) (Pipe, error)

InsertPipe inserts a row. sourceID nil = uncollared (symmetric many-to-many pipe under the manifold); non-nil = collared under that source. The caller passes accountID explicitly because uncollared rows can't derive it from source.account_id. Retention overrides are NULL when the pointer arg is nil — the server provisions the JetStream stream with defaults.

func ListPipesForSource 

func ListPipesForSource(ctx context.Context, p *Pool, sourceID uuid.UUID) ([]Pipe, error)

ListPipesForSource returns the user-creatable pipes for one source, sorted by name. Excludes auto-provisioned pipes (those aren't stored).

func ListUncollaredPipesForAccount added in v0.31.0 

func ListUncollaredPipesForAccount(ctx context.Context, p *Pool, accountID uuid.UUID) ([]Pipe, error)

ListUncollaredPipesForAccount returns every uncollared pipe row in the account, sorted (manifold, name). Used by `ppz ls` to surface the sourceless rows that walking sources alone misses. Phase 1.5.

type Pool 

type Pool struct {
	*pgxpool.Pool
}

Pool is the public handle other packages use. It wraps pgxpool.Pool so the rest of the codebase doesn't import pgx directly.

func Open 

func Open(ctx context.Context, url string) (*Pool, error)

type Schedule added in v0.51.0 

type Schedule struct {
	ID              uuid.UUID
	AccountID       uuid.UUID
	Manifold        string
	SourceHandle    string
	Pipe            string
	Payload         string
	Sender          string
	Kind            string
	Spec            string
	TZ              string
	NextFireAt      time.Time
	LastFiredAt     *time.Time
	FailCount       int // consecutive failed fires; reset on success
	CreatedByUserID uuid.UUID
	CreatedAt       time.Time
	CreatorUsername string
}

Schedule is one live scheduled send. SourceHandle=="" targets an uncollared pipe at Manifold. Spec is display-shaped (RFC3339 for at, Go duration for every, cron expression for cron); CreatedAt anchors the interval grid for kind=every. LastFiredAt is nil until the first fire. CreatorUsername is join-filled by ListSchedules for the `schedule ls` CREATOR column (not a table column).

func ClaimDueSchedules added in v0.51.0 

func ClaimDueSchedules(ctx context.Context, p *Pool, now time.Time, limit int) ([]Schedule, error)

ClaimDueSchedules atomically claims up to limit due rows (next_fire_at <= now): SKIP LOCKED against concurrent claimers, lease-bumped so the claim survives past the transaction. Returns the rows with their PRE-claim next_fire_at — the value the missfire policy (schedule.Decide) judges lateness against.

func InsertSchedule added in v0.51.0 

func InsertSchedule(ctx context.Context, p *Pool, s Schedule) (Schedule, error)

InsertSchedule stores a new schedule. ID is stamped here; CreatedAt is honoured when set (it anchors kind=every's interval grid, so the handler passes the same instant it computed next_fire_at from) and stamped otherwise.

The CLI addresses schedules by ShortID (last 8 hex of the uuid), so a per-account suffix collision would make `schedule ls`/`rm` ambiguous. Astronomically unlikely (~n²/2³³), but free to prevent: regenerate the uuid when the suffix is already taken in the account.

func ListSchedules added in v0.51.0 

func ListSchedules(ctx context.Context, p *Pool, accountID uuid.UUID) ([]Schedule, error)

ListSchedules returns one account's live schedules, soonest next_fire_at first, with CreatorUsername join-filled.

func (Schedule) ShortID added in v0.51.0 

func (s Schedule) ShortID() string

ShortID returns the CLI-facing 8-char id: the last 8 hex of the row's uuid, matching `sent id=…`'s lastHex8 convention.

type Source 

type Source struct {
	ID                   uuid.UUID
	AccountID            uuid.UUID
	CreatedByUserID      uuid.UUID // user that created the source (NOT NULL)
	Manifold             string    // hierarchical-grouping segment; ” = root (NOT NULL on the DB column)
	Handle               string
	Kind                 SourceKind
	CreatedAt            time.Time
	LastBroadcastAt      *time.Time
	LastBroadcastPayload *string
}

func GetSourceByHandle 

func GetSourceByHandle(ctx context.Context, p *Pool, accountID uuid.UUID, handle string) (Source, error)

func InsertSource 

func InsertSource(ctx context.Context, p *Pool, accountID, createdBy uuid.UUID, manifold, handle string, kind SourceKind) (Source, error)

InsertSource creates a row attributed to `createdBy` (NOT NULL on the table). Server callers stamp this from the API-key's CreatedByUserID (API path) or caller.UserID (OAuth path).

func ListSourcesForOrg 

func ListSourcesForOrg(ctx context.Context, p *Pool, accountID uuid.UUID) ([]Source, error)

func (Source) IsAutoPipe added in v0.22.1 

func (s Source) IsAutoPipe(name string) bool

IsAutoPipe reports whether name is an auto-provisioned pipe for this source kind (i.e. JetStream-only, not stored in the pipes table).

func (Source) Pipes 

func (s Source) Pipes() []string

Pipes returns the pipe set for a source based on its kind. All sources have:

  • inbox: direct messages intended for this source/agent

pty sources also have:

  • stdin: input fed to the wrapped child via `ppz send`
  • stdout: byte-faithful capture of the PTY master's output (ANSI escapes intact); both `ppz read` and `ppz terminal view` consume this pipe.
  • stdctrl: control plane (resize events, etc.).

`broadcast` was an auto-provisioned pipe pre-launch; it was removed in Phase 1 (locked decision #16) — teams now use explicit room pipes (`ppz pipe create team1.room` with --writers=anyone) for shared channels.

type SourceKind 

type SourceKind string

SourceKind enumerates the supported source shapes. 

"message" — default; two pipes: broadcast, inbox.
"pty"     — terminal source; broadcast, inbox, and terminal IO pipes.

const (
	SourceKindMessage SourceKind = "message"
	SourceKindPTY     SourceKind = "pty"
)

type User 

type User struct {
	ID        uuid.UUID
	Username  string
	Email     string
	Mode      UserMode
	GitHubID  *int64 // nil for mode=internal users
	AvatarURL string
	CreatedAt time.Time
}

func GetUser 

func GetUser(ctx context.Context, p *Pool, id uuid.UUID) (User, error)

func GetUserByUsername 

func GetUserByUsername(ctx context.Context, p *Pool, username string) (User, error)

func InsertUser 

func InsertUser(ctx context.Context, p *Pool, username, email string, mode UserMode) (User, error)

func ListMembers 

func ListMembers(ctx context.Context, p *Pool, accountID uuid.UUID) ([]User, error)

ListMembers returns the non-owner members of an org, ordered by when they were added.

func UpsertUserByGitHubID 

func UpsertUserByGitHubID(ctx context.Context, p *Pool, githubID int64, username, email, avatarURL string) (User, bool, error)

UpsertUserByGitHubID inserts a brand new mode=github user, or updates the existing row matching the GitHub numeric id. Returns the resolved User row plus a bool indicating whether the row was freshly created (true) vs already existed (false). Callers use the bool to decide whether to auto-create the user's first org.

type UserMode 

type UserMode string

UserMode = "github" (real OAuth identity) or "internal" (placeholder / seeded test user / pre-OAuth-era account). 

const (
	UserModeGithub   UserMode = "github"
	UserModeInternal UserMode = "internal"
)

Source Files

View all Source files

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.