drive

package
v0.2.30 Latest Latest
Warning

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

 Go to latest Published: Sep 16, 2026 License: MPL-2.0 Imports: 21 Imported by: 0

Documentation

Overview

Package drive is the mutable document store behind the chassis `webdav` personality and the txco://drive/* ops: per-tenant collections of files and directories whose bytes live in an ordinary object store and whose names, versions and hierarchy live in a SQL index.

It is the third storage plane beside the stack's FILES/ (immutable, content-addressed, served) and the blob layer (named content over the same CAS). A drive resource is MUTABLE: PUT replaces, MOVE renames, DELETE deletes. What stays fixed is the resource id — `dr_<hxid>`, minted once when the resource is created and carried through every rename and rewrite. The path is where the resource is; the etag (sha256 of the current bytes) is which version it is. A consumer that keys on the resource id survives a rename; one that keys on the etag sees every content change.

Bytes first, metadata second (the house rule): a put streams the body to a fresh object key, then opens the index transaction; an object without a live row is an orphan the sweeper reclaims, never a row without bytes. Deletes leave a tombstone (deleted_at) with a bumped modseq so a `list since=<modseq>` can report removals; the partial unique index on (collection_id, path) WHERE deleted_at IS NULL frees the path at once, and the sweeper hard-deletes tombstones past a retention window.

Every mutation advances the collection's sync_token in the same transaction and reports itself to the configured MutationSink; with a transactional sink (the scheduled store on the same database) the event row commits with the mutation.

Storage is dialect-aware (registry.Dialect, the seam auth, scheduled, calendar and contacts use); the bundled index backend is a SQLite file of its own, never the runtime DB.

Index

Constants

View Source 
const (
	MaxPathBytes    = 1024
	MaxPathSegments = 32
	MaxSegmentBytes = 255
)

Path limits. A path is the "/"-joined segments below the collection root; the root itself is the empty path.

View Source 
const (
	EventResourceCreated = "drive.resource.created"
	EventResourceUpdated = "drive.resource.updated"
	EventResourceDeleted = "drive.resource.deleted"
	EventResourceMoved   = "drive.resource.moved"
)

Mutation event names: `drive.resource.<verb>`, one per committed mutation of a resource. A subtree operation (move, copy or delete of a directory) reports ONE event, for the directory; descendants are re-derived by listing.

View Source 
const (
	StatusActive   = "active"
	StatusDisabled = "disabled"
)

Account statuses.

View Source 
const (
	KindFile = "file"
	KindDir  = "dir"
)

Resource kinds.

Variables

View Source 
var (
	// ErrNotFound is a missing collection, account or resource.
	ErrNotFound = errors.New("drive: not found")
	// ErrExists is a create that found a live resource at the path (mkdir,
	// or a move/copy without overwrite).
	ErrExists = errors.New("drive: resource exists")
	// ErrNoParent is a write whose parent directory does not exist.
	ErrNoParent = errors.New("drive: parent directory does not exist")
	// ErrNotDirectory is a directory operation on a file.
	ErrNotDirectory = errors.New("drive: not a directory")
	// ErrIsDirectory is a file operation on a directory.
	ErrIsDirectory = errors.New("drive: is a directory")
	// ErrPrecondition is a failed If-Match / If-None-Match.
	ErrPrecondition = errors.New("drive: precondition failed")
	// ErrQuota is a write that would exceed the collection's byte or
	// resource budget.
	ErrQuota = errors.New("drive: collection quota exceeded")
	// ErrTooLarge is a file over the per-file byte cap.
	ErrTooLarge = errors.New("drive: file too large")
	// ErrSizeMismatch is a put whose body was shorter or longer than its
	// declared size.
	ErrSizeMismatch = errors.New("drive: body size does not match declared size")
	// ErrBadPath is a path that fails NormalizePath, or the root where a
	// resource path is required.
	ErrBadPath = errors.New("drive: invalid path")
	// ErrCycle is a move/copy whose destination is the source or inside it.
	ErrCycle = errors.New("drive: destination is the source or inside it")
	// ErrUsernameTaken is returned by UpsertAccount when the username already
	// belongs to another tenant (usernames are globally unique: they are the
	// login identity).
	ErrUsernameTaken = errors.New("drive: username belongs to another tenant")
	// ErrDisabled is an operation against a disabled account.
	ErrDisabled = errors.New("drive: account disabled")
	// ErrNotEmpty is a collection delete that found live resources and was
	// not forced.
	ErrNotEmpty = errors.New("drive: collection is not empty")
)

Functions

func BaseOf 

func BaseOf(p string) string

BaseOf returns the last segment of a normalized path ("" for the root).

func DefaultContentType 

func DefaultContentType(p string) string

DefaultContentType derives a content type from a path's extension, application/octet-stream when none is known.

func Depth 

func Depth(p string) int

Depth is the number of segments of a normalized path (0 for the root).

func ETagOf 

func ETagOf(b []byte) string

ETagOf is the etag of stored bytes: bare sha256 hex.

func IdempotencyKey 

func IdempotencyKey(m Mutation) string

IdempotencyKey is the mutation's de-duplication key for a queue that coalesces on it: `drive:<resource_id>:<modseq>`. Never the resource id alone — the scheduled store overwrites a still-pending row on key conflict, and two quick writes to one resource must both be delivered.

func IsInside 

func IsInside(p, dir string) bool

IsInside reports whether p is strictly below dir (both normalized). The root ("") contains every non-root path.

func JoinPath 

func JoinPath(parent, seg string) string

JoinPath joins a normalized parent and a segment.

func LikePrefix 

func LikePrefix(p string) string

LikePrefix is the LIKE pattern matching every path strictly below p, with LIKE metacharacters escaped; the SQL that uses it must say `ESCAPE '\'` (likeEscape).

func NewResourceID 

func NewResourceID() string

NewResourceID mints a resource id (`dr_<hxid>`, time-sortable).

func NormalizePath 

func NormalizePath(p string) (string, error)

NormalizePath canonicalizes a resource path: NFC (so "é" typed two ways is one name — macOS clients send NFD), leading/trailing "/" trimmed, each segment non-empty, not "." or "..", valid UTF-8, no NUL and no "/" (by construction), within the byte and depth limits. Returns "" for the root ("", "/", "//"). Rejects with ErrBadPath (wrapped, with the reason).

func NormalizeUsername 

func NormalizeUsername(u string) string

NormalizeUsername lowercases and trims a login / op username.

func ParentOf 

func ParentOf(p string) string

ParentOf returns the parent path of a normalized path ("" for a top-level resource or the root).

func Register 

func Register(name string, c Constructor)

Register adds an index backend constructor. Called from a backend package's init().

func RegisterObjects 

func RegisterObjects(name string, c ObjectsConstructor)

RegisterObjects adds an object backend constructor. Called from a backend package's init(); the chassis activates a backend with a blank import.

func ValidCollectionName 

func ValidCollectionName(name string) bool

ValidCollectionName reports whether name is a collection name: a URL segment of up to 128 chars (ops use short lowercase names; a pony's collection is its slug).

func ValidKeySegment 

func ValidKeySegment(seg string) bool

ValidKeySegment reports whether one "/"-separated object-key segment is safe for any backend: `[A-Za-z0-9._-]`, non-empty, not only dots (the filesystem traversal entries). Keys are built from ids the Store mints, so this is a reject check, not a sanitizer — a bad segment is a bug.

Types

type Account 

type Account struct {
	Tenant       string
	Username     string
	PwHash       string
	Status       string
	CollectionID string
	CreatedAt    time.Time
	UpdatedAt    time.Time
}

Account is a drive_accounts row: a Basic-auth login for the `webdav` head, bound to exactly ONE collection (the DAV root the login sees).

type Collection 

type Collection struct {
	ID            string
	Tenant        string
	Name          string
	SyncToken     int64
	BytesUsed     int64
	ResourceCount int64
	CreatedAt     time.Time
	UpdatedAt     time.Time
}

Collection is a live drive_collections row.

type Config 

type Config struct {
	// DBPath is the bundled SQLite backend's file path (--drive-db-path).
	DBPath string
	// Objects is the object store the index points into (--drive-objects).
	Objects ObjectStore
}

Config carries backend-selecting options resolved from chassis config for the INDEX backend. The bundled "sqlite" backend uses DBPath; every backend receives the already-opened object store to hold the bytes. Adding a field here doesn't affect existing backends (same posture as contacts.Config).

type Constructor 

type Constructor func(Config) (*Store, error)

Constructor builds a Store from resolved config. It is expected to open its backing DB and call Store.EnsureSchema before returning.

type DeleteOpts 

type DeleteOpts struct {
	// IfMatch is the client's conditional header, unquoted ("*" = any).
	IfMatch string
}

DeleteOpts steers Delete.

type Limits 

type Limits struct {
	// MaxFileBytes caps one file's size.
	MaxFileBytes int64
	// MaxCollectionBytes caps a collection's bytes_used.
	MaxCollectionBytes int64
	// MaxResources caps a collection's live resource_count.
	MaxResources int64
}

Limits are the write-side budgets a Store enforces. Zero means unlimited.

type ListOpts 

type ListOpts struct {
	// Path is the directory whose entries are listed ("" = the root).
	Path string
	// Recursive lists every resource below Path instead of its direct
	// children.
	Recursive bool
	// SinceModSeq returns only rows changed after that sync token.
	SinceModSeq int64
	// IncludeDeleted returns tombstones too (a sync consumer's deletions).
	IncludeDeleted bool
	// Limit caps the page (0 = unlimited); After is the exclusive path
	// cursor of the previous page.
	Limit int
	After string
}

ListOpts narrows List.

type Mutation 

type Mutation struct {
	Event        string
	Tenant       string
	CollectionID string
	Collection   string // the collection's name
	ResourceID   string
	Kind         string
	Path         string
	FromPath     string // moved: the previous path
	ETag         string
	Size         int64
	ContentType  string
	ModSeq       int64
	At           time.Time
}

Mutation is the fact a committed mutation reports to the sink: enough for a consumer to decide whether to fetch (kind, size, content_type, etag) and to find the bytes (collection + resource_id), without the bytes themselves riding the event.

type MutationSink 

type MutationSink interface {
	Enqueue(ctx context.Context, m Mutation) error
}

MutationSink receives every committed mutation. Enqueue is called AFTER the index transaction committed; an error is reported to the store's sink-error handler and does not undo the mutation.

type NopSink 

type NopSink struct{}

NopSink drops every mutation (the default; a node without a scheduled store).

func (NopSink) Enqueue 

func (NopSink) Enqueue(context.Context, Mutation) error

Enqueue implements MutationSink.

type ObjectCopier 

type ObjectCopier interface {
	Copy(ctx context.Context, from, to string) error
}

ObjectCopier is the optional server-side copy capability (S3 CopyObject, a hard link). The Store type-asserts it; a backend without it is copied through Get → Put.

type ObjectInfo 

type ObjectInfo struct {
	Size    int64
	ModTime time.Time
	// ContentType is what the backend recorded at Put, when it records one
	// (S3 does; the file backend does not). The index is authoritative.
	ContentType string
}

ObjectInfo is what an object store knows about a stored object without reading it.

type ObjectStore 

type ObjectStore interface {
	// Put streams r to key. size is the caller's declared byte count when
	// known (-1 otherwise) — a hint a backend may use to pick a single-part
	// upload; the Store verifies the count itself. A Put must be atomic: a
	// reader never observes a partial object, and a failed Put leaves no
	// object at key.
	Put(ctx context.Context, key string, r io.Reader, size int64, contentType string) error

	// Get opens the object for reading. ErrNotFound if absent.
	Get(ctx context.Context, key string) (io.ReadCloser, ObjectInfo, error)

	// Stat reports the object's metadata. ErrNotFound if absent.
	Stat(ctx context.Context, key string) (ObjectInfo, error)

	// List walks every object under prefix ("" for all), calling fn for each;
	// fn's error stops the walk and is returned. Pagination is the backend's.
	List(ctx context.Context, prefix string, fn func(key string, info ObjectInfo) error) error

	// Delete removes key. Absent key is not an error (idempotent).
	Delete(ctx context.Context, key string) error

	// Name is the backend identity ("file", "s3").
	Name() string
}

ObjectStore holds resource bytes under keys the Store mints: `<tenant>/<collection_id>/<resource_id>/<version_id>` — "/"-separated, every segment `[A-Za-z0-9._-]`. A backend adds its own root (a directory, an S3 prefix) and must never let a key escape it. Objects are written once and never rewritten: a new version is a new key, and the sweeper deletes keys no live index row references.

The interface is deliberately minimal (put / get / stat / list / delete): it is what a mutable document store needs from ordinary object storage, and what a fleet overlay can implement over S3 in one file.

func OpenObjects 

func OpenObjects(name string, cfg ObjectsConfig) (ObjectStore, error)

OpenObjects constructs the named object backend. Unknown name is a startup error listing what is available.

type ObjectsConfig 

type ObjectsConfig struct {
	// FileDir is the bundled file backend's root (--drive-objects-file-dir).
	FileDir string
}

ObjectsConfig carries backend-selecting options for the OBJECT store (--drive-objects). The bundled "file" backend uses FileDir; an overlay's S3 backend reads its own environment.

type ObjectsConstructor 

type ObjectsConstructor func(ObjectsConfig) (ObjectStore, error)

ObjectsConstructor builds an ObjectStore from resolved config.

type PutOpts 

type PutOpts struct {
	// IfMatch / IfNoneMatch are the client's conditional headers, unquoted
	// ("*" for IfNoneMatch means create-only; "*" for IfMatch means
	// must-exist).
	IfMatch     string
	IfNoneMatch string
	// ContentType is recorded on the row; empty keeps the existing one on
	// an update and derives one from the extension on a create.
	ContentType string
}

PutOpts steers Put.

type PutResult 

type PutResult struct {
	ResourceID string
	ETag       string
	Size       int64
	Created    bool
	Noop       bool
	ModSeq     int64
}

PutResult reports what Put did.

type Resource 

type Resource struct {
	ResourceID   string
	CollectionID string
	Path         string
	ParentPath   string
	Depth        int
	Kind         string
	ObjectKey    string
	Size         int64
	ContentType  string
	ETag         string
	ModSeq       int64
	CreatedAt    time.Time
	UpdatedAt    time.Time
	DeletedAt    time.Time
	Deleted      bool
}

Resource is a drive_resources row: one file or directory of a collection. ResourceID is the identity (fixed for life); Path is the location (changes on move); ETag is the version (sha256 of the current bytes, "" for a directory); ModSeq is the collection sync_token value at the row's last change.

func (Resource) IsDir 

func (r Resource) IsDir() bool

IsDir reports a directory resource.

type Store 

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

Store is the façade over the index tables plus the object store that holds the bytes. It carries the dialect (for `?`→`$n` rebinding), a clock seam for tests, the optional mutation sink and the write limits.

func NewStore 

func NewStore(db *sql.DB, d registry.Dialect, objects ObjectStore) *Store

NewStore builds a Store over the opened index DB, its dialect and the object store for bytes. A nil dialect defaults to SQLite (the in-tree default). The sink defaults to NopSink.

func Open 

func Open(name string, cfg Config) (*Store, error)

Open constructs the named index backend. Unknown name is a startup error listing what is available (so a misconfigured --drive-store fails loudly).

func (*Store) Close 

func (s *Store) Close() error

Close releases the underlying DB handle. The object store is the caller's to close.

func (*Store) Copy 

func (s *Store) Copy(ctx context.Context, collID, from, to string, overwrite bool) (Resource, error)

Copy duplicates the resource at from (a directory with everything below it) to to. Every copy gets a new resource id; the bytes are copied server-side when the object store can (ObjectCopier), else streamed. Bytes first: the objects are duplicated from a snapshot of the source rows, then the index transaction re-reads the source and inserts the rows — a source row that changed in between is copied again inline; a failed transaction leaves orphans the sweeper reclaims. ErrCycle / ErrNoParent / ErrExists as Move. One event, for the copied root.

func (*Store) DB 

func (s *Store) DB() *sql.DB

DB exposes the index handle for tests and backends. Not for request paths.

func (*Store) Delete 

func (s *Store) Delete(ctx context.Context, collID, p string, opts DeleteOpts) (Resource, error)

Delete tombstones the resource at p (a directory with everything below it). ErrNotFound when absent.

func (*Store) DeleteByID 

func (s *Store) DeleteByID(ctx context.Context, collID, resID string, opts DeleteOpts) (Resource, error)

DeleteByID is Delete addressed by resource id.

func (*Store) DeleteCollection 

func (s *Store) DeleteCollection(ctx context.Context, tenant, name string, force bool) (bool, error)

DeleteCollection soft-deletes the tenant's collection of that name and tombstones its live resources. Refuses (ErrNotEmpty) when the collection holds live resources unless force. Returns false (no error) when there is no such live collection. No mutation events are emitted: the collection itself is the subscription a consumer keys on.

func (*Store) EnsureCollection 

func (s *Store) EnsureCollection(ctx context.Context, tenant, name string) (Collection, bool, error)

EnsureCollection returns the tenant's live collection of that name, creating it when absent. A soft-deleted collection of the same name is NOT resurrected — its resources are tombstoned under the old id and a new collection starts empty. created reports a fresh row.

func (*Store) EnsureSchema 

func (s *Store) EnsureSchema(ctx context.Context) error

EnsureSchema creates the tables + indexes if absent. Portable DDL: TEXT ids (hxid), TEXT RFC3339 timestamps, native partial indexes, BIGINT counters — one DDL serves SQLite and Postgres. Additive-only: a later column is added with its own IF-NOT-EXISTS-shaped statement.

func (*Store) GetAccount 

func (s *Store) GetAccount(ctx context.Context, username string) (Account, bool, error)

GetAccount looks an account up by username (the login identity).

func (*Store) GetCollection 

func (s *Store) GetCollection(ctx context.Context, tenant, name string) (Collection, bool, error)

GetCollection returns the tenant's live collection of that name.

func (*Store) GetCollectionByID 

func (s *Store) GetCollectionByID(ctx context.Context, id string) (Collection, bool, error)

GetCollectionByID returns a live collection by id.

func (*Store) Limits 

func (s *Store) Limits() Limits

Limits reports the write budgets.

func (*Store) List 

func (s *Store) List(ctx context.Context, collID string, opts ListOpts) ([]Resource, error)

List returns resources per opts, ordered by path. Without Recursive the direct children of Path; with it every resource strictly below Path. The directory itself is never included. Whether Path exists is the caller's question (Stat).

func (*Store) ListCollections 

func (s *Store) ListCollections(ctx context.Context, tenant string) ([]Collection, error)

ListCollections returns the tenant's live collections by name.

func (*Store) Mkdir 

func (s *Store) Mkdir(ctx context.Context, collID, p string) (Resource, error)

Mkdir creates a directory at p. ErrExists when a live resource of either kind is there; ErrNoParent when the parent is missing.

func (*Store) Move 

func (s *Store) Move(ctx context.Context, collID, from, to string, overwrite bool) (Resource, error)

Move renames the resource at from to to, keeping its resource id (and, for a directory, every descendant's). ErrCycle when to is from or inside it; ErrNoParent when to's parent is missing; ErrExists when to is live and overwrite is false — with overwrite the destination subtree is deleted first. One event, for the moved root.

func (*Store) Objects 

func (s *Store) Objects() ObjectStore

Objects exposes the object store (the sweeper and tests).

func (*Store) Open 

func (s *Store) Open(ctx context.Context, collID, p string) (io.ReadCloser, Resource, error)

Open returns a reader over the file at p and its row. ErrNotFound when absent; ErrIsDirectory for a directory (the root included).

func (*Store) OpenByID 

func (s *Store) OpenByID(ctx context.Context, collID, resID string) (io.ReadCloser, Resource, error)

OpenByID is Open addressed by resource id.

func (*Store) Put 

func (s *Store) Put(ctx context.Context, collID, p string, r io.Reader, size int64, opts PutOpts) (PutResult, error)

Put stores size bytes from r at path p in the collection, creating or replacing the file. Bytes first: the body streams to a fresh object key (hashed on the way), then one index transaction locks the collection, re-checks parent / kind / preconditions / quota, writes the row, advances the sync token and reports the mutation. An unchanged body (same etag) is a Noop: no version bump, no event, and the freshly written object is discarded. A directory at p is ErrIsDirectory; a missing parent ErrNoParent; a body shorter or longer than size ErrSizeMismatch.

func (*Store) SetClock 

func (s *Store) SetClock(now func() time.Time)

SetClock pins the store's clock (tests).

func (*Store) SetLimits 

func (s *Store) SetLimits(l Limits)

SetLimits installs the write budgets.

func (*Store) SetSink 

func (s *Store) SetSink(sink MutationSink)

SetSink installs the mutation sink. A sink that also implements TransactionalMutationSink is called inside the index transaction; any other sink is called after commit.

func (*Store) SetSinkErrorHandler 

func (s *Store) SetSinkErrorHandler(fn func(Mutation, error))

SetSinkErrorHandler installs the hook told when a post-commit sink call fails (boot wires it to the logger).

func (*Store) Stat 

func (s *Store) Stat(ctx context.Context, collID, p string) (Resource, bool, error)

Stat returns the live resource at p. The root ("") is a synthetic directory carrying the collection's sync token.

func (*Store) StatByID 

func (s *Store) StatByID(ctx context.Context, collID, resID string) (Resource, bool, error)

StatByID returns the live resource with that id in the collection.

func (*Store) Sweep 

func (s *Store) Sweep(ctx context.Context, grace, tombstoneRetention time.Duration) (SweepReport, error)

Sweep is the drive janitor, idempotent and safe to repeat or to run on several nodes at once:

  1. Hard-delete tombstoned resources (and collections) whose deleted_at is older than tombstoneRetention — after that a `list since` consumer can no longer learn of the deletion, which is the window it is given.
  2. Walk the object store and delete every object older than grace that no LIVE row references: superseded versions after a rewrite, the objects of hard-deleted tombstones, and the leftovers of a put or copy whose index transaction failed.

grace protects an in-flight put (bytes land before the row) and an in-flight reader of a version that was just superseded: nothing younger than grace is touched. Per-object errors are counted and skipped.

func (*Store) UpsertAccount 

func (s *Store) UpsertAccount(ctx context.Context, tenant, username, pwHash, status, collectionID string) (created bool, err error)

UpsertAccount creates the account or updates it. An empty pwHash / status / collectionID leaves the stored value unchanged on update; pwHash and collectionID are required on create. created reports whether the row was new.

type SweepReport 

type SweepReport struct {
	// Objects walked in the object store.
	Scanned int
	// Objects deleted because no live row referenced them.
	Orphans int
	// Tombstoned resource rows hard-deleted past retention.
	Tombstones int64
	// Tombstoned collection rows hard-deleted past retention.
	Collections int64
	// Per-object errors logged and skipped (the pass continues).
	Errors int
	// Bytes the deleted orphans held.
	OrphanBytes int64
}

SweepReport counts what one Sweep pass did.

type TransactionalMutationSink 

type TransactionalMutationSink interface {
	MutationSink
	EnqueueTx(ctx context.Context, tx *sql.Tx, m Mutation) error
}

TransactionalMutationSink is a sink that can write inside the index transaction (the scheduled store on the same database). The store prefers EnqueueTx when the sink implements it: the event row then commits with the mutation or not at all.

Source Files

View all Source files

Directories

Path Synopsis
Package filestore is the local-filesystem drive object backend: the bundled default, zero infrastructure, directly inspectable.
 Package filestore is the local-filesystem drive object backend: the bundled default, zero infrastructure, directly inspectable.
Package scheduledsink delivers drive mutation events through the scheduled store: every committed mutation becomes one pending scheduled_events row, due now, that the `scheduled` personality fires into the tenant's `_scheduled/0` stack like any other scheduled event (`@src == "scheduled"`, the facts under `@scheduled.payload`).
 Package scheduledsink delivers drive mutation events through the scheduled store: every committed mutation becomes one pending scheduled_events row, due now, that the `scheduled` personality fires into the tenant's `_scheduled/0` stack like any other scheduled event (`@src == "scheduled"`, the facts under `@scheduled.payload`).

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.