Documentation
¶
Overview ¶
Package files is the service / use-case layer. It owns the write path and the consistency logic between storage and metadata; layers above it (api) talk only to this package and never reach storage or metadata directly.
Index ¶
- Constants
- Variables
- type ArchiveEntry
- type ArchiveLimits
- type EntryScope
- type ExtractInProgressError
- type ExtractOptions
- type ExtractSummary
- type File
- type Job
- type JobStatus
- type ListFilter
- type Option
- type Repository
- type Service
- func (s *Service) ArchiveEntries(ctx context.Context, id string) (*File, []ArchiveEntry, error)
- func (s *Service) CancelExtraction(ctx context.Context, jobID string) (*Job, error)
- func (s *Service) Delete(ctx context.Context, id string) error
- func (s *Service) Extract(ctx context.Context, id string, o ExtractOptions) (*ExtractSummary, error)
- func (s *Service) ExtractionJob(ctx context.Context, jobID string) (*Job, error)
- func (s *Service) List(ctx context.Context, filter ListFilter) ([]*File, error)
- func (s *Service) Metadata(ctx context.Context, id string) (*File, error)
- func (s *Service) Open(ctx context.Context, id string) (*File, io.ReadCloser, error)
- func (s *Service) OpenRange(ctx context.Context, f *File, offset, length int64) (io.ReadCloser, error)
- func (s *Service) Presign(ctx context.Context, id string, ttl time.Duration) (*File, string, error)
- func (s *Service) SubmitExtraction(ctx context.Context, id string, selected []int) (*Job, error)
- func (s *Service) UpdateMetadata(ctx context.Context, id string, tags map[string]any, merge bool) (*File, error)
- func (s *Service) Upload(ctx context.Context, in UploadInput) (*File, error)
- func (s *Service) UploadIdempotent(ctx context.Context, key string, in UploadInput) (f *File, replayed bool, err error)
- type SkipReason
- type SkippedEntry
- type UploadInput
Constants ¶
const ( JobPending = metadata.JobPending JobRunning = metadata.JobRunning JobSucceeded = metadata.JobSucceeded JobFailed = metadata.JobFailed JobCancelled = metadata.JobCancelled )
Job lifecycle states.
const ( EntriesInclude = metadata.EntriesInclude EntriesExclude = metadata.EntriesExclude EntriesOnly = metadata.EntriesOnly )
List scopes, re-exported for the api layer.
const ReservedTagPrefix = metadata.ReservedTagPrefix
ReservedTagPrefix marks metadata keys the service writes and clients may not: the api layer rejects them on upload and PATCH, and the repository carries them across a replace.
Variables ¶
var ErrArchiveTooLarge = errors.New("files: archive exceeds the extraction limits")
ErrArchiveTooLarge means the archive, or the selection to extract from it, exceeds a configured cap. The api layer maps it to 413; the wrapped message names the cap and the numbers.
var ErrChecksumMismatch = errors.New("files: checksum mismatch")
ErrChecksumMismatch is surfaced by a verifying download reader when the stored bytes no longer hash to the recorded checksum (Phase 2 integrity).
var ErrExtractCancelled = errors.New("files: extraction cancelled")
ErrExtractCancelled is returned by Extract when its progress hook asked it to stop. The partial summary is returned alongside it: the children written so far remain, and a later run resumes past them.
var ErrExtractInProgress = errors.New("files: an extraction of this archive is already in progress")
ErrExtractInProgress means another extraction job for the same archive is still pending or running. The api layer maps it to 409; the value returned is an *ExtractInProgressError naming that job, which is the one to poll.
var ErrIdempotencyConflict = errors.New("files: idempotency key conflict")
ErrIdempotencyConflict means an upload with the same Idempotency-Key is still in progress. The api layer maps it to 409 — the client should retry shortly.
var ErrIdempotencyResultGone = errors.New("files: idempotent result deleted")
ErrIdempotencyResultGone means the file an Idempotency-Key produced has since been deleted. Retrying under the same key can never succeed until the key expires, so the api layer maps it to 410 — the client must pick a new key.
var ErrInvalidID = errors.New("files: invalid id")
ErrInvalidID is returned when an id is not a well-formed file identifier. The api layer maps it to 400 — a malformed id is a client error, not a 500.
var ErrInvalidSelection = errors.New("files: entry selection out of range")
ErrInvalidSelection means a requested entry index is outside the archive's directory. The api layer maps it to 400.
var ErrJobNotFound = errors.New("files: extraction job not found")
ErrJobNotFound is returned when no extraction job matches — including one the retention sweep has already purged. The api layer maps it to 404.
var ErrNotArchive = errors.New("files: not a supported archive (zip)")
ErrNotArchive means the file the archive endpoints were pointed at is not a supported archive. The api layer maps it to 415 — the one status a caller will hit by accident, so it is named rather than left to a generic 400.
var ErrNotFound = errors.New("files: not found")
ErrNotFound is returned when no live file matches. It is the api layer's only signal for a 404, keeping storage/metadata sentinels below this seam.
Functions ¶
This section is empty.
Types ¶
type ArchiveEntry ¶ added in v1.3.0
ArchiveEntry and SkipReason are aliased so the api layer can describe an archive without importing the archive package directly.
type ArchiveLimits ¶ added in v1.3.0
type ArchiveLimits struct {
MaxEntryBytes int64 // per extracted entry — an entry is an upload, so this reuses the upload cap
MaxEntries int // entries in the archive's directory; both endpoints refuse larger archives
MaxTotalBytes int64 // uncompressed bytes one extraction may write
MaxRatio int64 // per-entry uncompressed ÷ compressed
Extensions []string // allowlist, lower-case with the leading dot; empty allows all
// MaxDirectoryBytes bounds the central directory's span, refusing an
// archive before its directory is parsed. It is the only cap applied
// pre-parse — MaxEntries cannot be, since the count does not exist until
// the parse is done. Zero does not disable it: it derives the bound from
// the entry ceiling instead (see maxRecordBudget).
MaxDirectoryBytes int64
}
ArchiveLimits bound an extraction. Zero disables each cap.
type EntryScope ¶ added in v1.3.0
type EntryScope = metadata.EntryScope
EntryScope and its values select extracted archive entries in a List.
type ExtractInProgressError ¶ added in v1.3.0
type ExtractInProgressError struct {
JobID string // may be empty if that job finished between the refusal and the lookup
}
ExtractInProgressError is ErrExtractInProgress carrying the id of the job that holds the archive, so the api layer can tell the caller which job to poll rather than only when to try again.
func (*ExtractInProgressError) Error ¶ added in v1.3.0
func (e *ExtractInProgressError) Error() string
func (*ExtractInProgressError) Unwrap ¶ added in v1.3.0
func (e *ExtractInProgressError) Unwrap() error
Unwrap makes errors.Is(err, ErrExtractInProgress) true.
type ExtractOptions ¶ added in v1.3.0
type ExtractOptions struct {
// Entries restricts the run to those directory indexes (an index outside
// the directory is ErrInvalidSelection); nil means every selectable
// entry, and an empty, non-nil slice selects nothing.
Entries []int
// OnProgress, when set, is called once before the first entry with
// (0, total) and again after each entry. Returning true stops the run
// after the entry just written — Extract then returns the partial summary
// with ErrExtractCancelled. The hook is what lets a job report progress
// and honour cancellation without this package knowing what a job is.
OnProgress func(done, total int) (cancel bool)
}
ExtractOptions configures one extraction run.
type ExtractSummary ¶ added in v1.3.0
type ExtractSummary struct {
ArchiveID string `json:"archive_id"`
Entries int `json:"entries"` // entries in the archive's directory
Extracted int `json:"extracted"`
Skipped int `json:"skipped"`
SkippedByReason map[SkipReason]int `json:"skipped_by_reason"`
SampleSkipped []SkippedEntry `json:"sample_skipped"` // the first sampleSkippedMax skipped entries
}
ExtractSummary is the bounded report of one extraction. It never lists the files it created: a large archive's results could not fit one response, so callers enumerate the children with a tag filter on the archive id instead.
The JSON tags are the wire shape — the api layer serialises it as is, and a job stores the same bytes — so there is one shape, not two.
type ListFilter ¶
type ListFilter = metadata.ListFilter
ListFilter narrows a List query. Aliased so the api layer can build filters without importing the metadata package directly.
type Option ¶ added in v1.3.0
type Option func(*Service)
Option configures a Service beyond its two collaborators.
func WithArchiveLimits ¶ added in v1.3.0
func WithArchiveLimits(l ArchiveLimits) Option
WithArchiveLimits sets the extraction caps. Without it extraction is uncapped.
type Repository ¶ added in v1.3.0
type Repository interface {
metadata.Repository
metadata.JobRepository
}
Repository is everything the service persists through: file records and extraction jobs. The two are separate seams in the metadata package; one pgx repository — and one in-memory double — satisfies both.
type Service ¶
type Service struct {
// contains filtered or unexported fields
}
Service orchestrates the object store and the metadata repository.
func New ¶
func New(backend storage.Backend, repo Repository, opts ...Option) *Service
New wires the service to its collaborators.
func (*Service) ArchiveEntries ¶ added in v1.3.0
ArchiveEntries lists the entries of the archive stored as id, each classified as selectable or skipped with a reason. It writes nothing, and it is what lets a caller answer "which file exactly?" before anything is stored.
func (*Service) CancelExtraction ¶ added in v1.3.0
CancelExtraction asks the job to stop and returns its state after the request. A pending job is cancelled at once; a running one stops after the entry it is writing, keeping the children written so far; a terminal one is unchanged — cancelling is idempotent and never an error on its own.
func (*Service) Delete ¶
Delete soft-deletes the file: the row is stamped deleted_at and immediately vanishes from reads (every read path filters deleted_at is null), while the object is purged asynchronously by GC (PLAN §7). Returns ErrNotFound if the file does not exist.
func (*Service) Extract ¶ added in v1.3.0
func (s *Service) Extract(ctx context.Context, id string, o ExtractOptions) (*ExtractSummary, error)
Extract stores every selected entry of the archive id as an ordinary file, tagged with the archive it came from.
A run is resumable by construction: entries that already have a child row are skipped as SkipAlreadyExtracted, so a run interrupted part-way — or simply repeated — completes the remainder instead of duplicating. A child the user deleted stays deleted (SkipPreviouslyDeleted). On any error, or a cancellation through OnProgress, the children written so far remain, and the next run picks up after them. Serialising concurrent runs on one archive is the job layer's concern: nothing here prevents two direct callers from writing the same archive at once.
func (*Service) ExtractionJob ¶ added in v1.3.0
ExtractionJob returns the job by id, or ErrJobNotFound — which a job past retention also answers, since the sweep deletes its row.
func (*Service) List ¶
List returns file records matching the filter, newest first. The metadata repository applies tag containment against the GIN index and keyset pagination by id. A malformed cursor is ErrInvalidID — it must be a file id handed out by a previous page.
func (*Service) Open ¶
Open streams the whole object through this process, verifying the SHA-256 as it reads: the returned reader yields ErrChecksumMismatch at EOF if the stored bytes no longer match the recorded checksum. The caller must Close it.
func (*Service) OpenRange ¶
func (s *Service) OpenRange(ctx context.Context, f *File, offset, length int64) (io.ReadCloser, error)
OpenRange streams a byte range [offset, offset+length) of an already-fetched record through this process — the caller resolves f via Metadata first (it needs the size to validate the range), so no second lookup happens here. Partial reads cannot be checksum-verified against the whole-object hash, so the reader is returned unverified. The caller must Close it.
func (*Service) Presign ¶
Presign returns the record and a time-limited URL that serves the object directly from the store, offloading the transfer from this process.
func (*Service) SubmitExtraction ¶ added in v1.3.0
SubmitExtraction enqueues an extraction of the archive id and returns the pending job. The run is validated first, exactly as a synchronous extraction would — the archive opens, the selection is in range, the total is under the cap — so every refusal a caller could act on is made now, with the same errors, rather than surfacing later as a failed job. A job for an archive that already has one pending or running is refused with an *ExtractInProgressError naming it: one active extraction per archive is a database invariant, and the caller's next move is to poll that job.
func (*Service) UpdateMetadata ¶
func (s *Service) UpdateMetadata(ctx context.Context, id string, tags map[string]any, merge bool) (*File, error)
UpdateMetadata writes tags onto a file's JSONB metadata, merging into the existing object or replacing it wholesale. Returns ErrNotFound if absent.
func (*Service) Upload ¶
Upload streams the body to storage, computing its SHA-256 inline, then commits the metadata row. On a metadata failure the freshly-written object is best-effort deleted; a leftover is reclaimed by GC.
func (*Service) UploadIdempotent ¶
func (s *Service) UploadIdempotent(ctx context.Context, key string, in UploadInput) (f *File, replayed bool, err error)
UploadIdempotent performs an upload guarded by a client-supplied idempotency key. The first call for a key uploads and records the result; a retry returns the original result (replayed=true) instead of creating a duplicate. A retry while the first is still in flight — or whose result has since been deleted — returns ErrIdempotencyConflict (the client should retry shortly).
On the replay/conflict paths the request body is intentionally not read; the HTTP server drains it.
type SkipReason ¶ added in v1.3.0
type SkipReason = archive.SkipReason
ArchiveEntry and SkipReason are aliased so the api layer can describe an archive without importing the archive package directly.
const ( SkipAlreadyExtracted SkipReason = "already_extracted" // a live child row exists for this index SkipPreviouslyDeleted SkipReason = "previously_deleted" // the child was soft-deleted: delete wins over re-extract SkipNotSelected SkipReason = "not_selected" // absent from the caller's entries list SkipCorrupt SkipReason = "corrupt" // the content failed the archive's own checks )
Skip reasons decided by the service rather than by the archive's directory.
type SkippedEntry ¶ added in v1.3.0
type SkippedEntry struct {
Index int `json:"index"`
Name string `json:"name"`
Reason SkipReason `json:"reason"`
}
SkippedEntry names one skipped entry and why.
Source Files