dataprivacy

package
v12.0.0 Latest Latest
Warning

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

 Go to latest Published: Aug 20, 2026 License: AGPL-3.0 Imports: 43 Imported by: 0

Documentation

Overview

Package dataprivacy fulfills GDPR and CCPA subject access and erasure requests as durable, auditable operations.

Subject access requests and erasure requests are legally mandatory, tedious, and structurally identical across applications: fan out over every domain that holds data about a subject, aggregate it, package it, deliver it safely, and record that you did. The application owns what its data is. This package owns doing it exactly once, durably, with an expiring artifact and an auditable result.

The registry, and the type that is not here

Adding a domain is a registration: 

registry := dataprivacy.NewRegistry()

if err := registry.RegisterCollector("identity", identityCollector); err != nil {
	return err
}
if err := registry.RegisterEraser("identity", identityEraser); err != nil {
	return err
}

A Collector returns already-encoded JSON — an opaque fragment the library never looks inside — and the library composes the fragments into a document by key.

That is the one place this deliberately departs from the prior art it generalizes. There, every domain wrote into a single shared aggregate struct, so adding a domain meant editing a central type that imported every domain package: 

// what this replaces
type UserDataCollection struct {
    Identity     identity.UserDataCollection
    MealPlanning mealplanning.UserDataCollection
    Webhooks     webhooks.UserDataCollection
    // ...eight more
}

A library cannot have that type — it would have to import its own consumers — and it turns out not to need one. The cost that type imposed was not hypothetical: it gained two fields in a single month, each an edit to the file most likely to conflict. It also meant one domain returning an error aborted the whole aggregate, so a subject's entire export failed because one unrelated table was slow. Fragments keyed by domain fix both: registration is local, and a failure is recorded against its key.

Collect and erase are separate interfaces

Erasure is not the inverse of export. Some data must be retained — financial records under tax law, audit entries under legitimate interest — and some must be anonymized in place rather than deleted, because a foreign key still points at it. Only the domain knows which of the three applies to each of its tables, so Eraser is registered separately and reports what it kept: 

func (e identityEraser) Erase(ctx context.Context, q database.SQLQueryExecutor, s dataprivacy.Subject) (dataprivacy.ErasureOutcome, error) {
    // ...
    return dataprivacy.ErasureOutcome{
        Deleted:    rows,
        Anonymized: anonymized,
        Retained: map[string]string{
            "invoices": "financial records, retained 7 years under [statute]",
        },
    }, nil
}

Partial exports are delivered; partial erasures are not

The two halves fail differently, and the asymmetry is deliberate.

Collection is isolated per key. A collector that errors or times out costs its own section: the artifact is still written, its manifest names the missing sections and why, and Request.Failures carries the same information. A subject with thirty days to complain is better served by most of their data plus an honest account of the gap than by nothing. An export in which *every* collector failed is a hard failure — a document asserting that nothing is held about a person is the one wrong answer available here.

A partial export is a successful operation, and that is worth being explicit about, because it is the one place this package's idea of success and the operations package's could have been made to disagree. The operation succeeded: it produced an artifact and a pointer to it. Which sections are missing is in ExportSummary, in the operation's Result.Detail, and in the artifact's own manifest — a fact about the answer rather than about whether there is one.

Erasure is atomic. Every registered Eraser shares one transaction with the request's own bookkeeping, so a subject is never left deleted from eight domains and present in three. A partial erasure has no coherent meaning and no status could describe it, so an eraser's error rolls the whole thing back and the request is retried intact.

The two operation kinds

Both halves run as operations. This package supplies two Runners and registers them; everything around them — the queue, the worker, the lease, the retry budget, the progress a client watches, the status endpoint — is the operations package's, shared with every other long-running thing in the application. 

KindExport   = "dataprivacy.export"
KindErasure  = "dataprivacy.erasure"

The registered domains are the progress denominator for free, because the registry already enumerates them: an export reports "3 of 9 domains complete" without a counting pass, and within a domain reports bytes collected, which is the untotalled counter a fan-out over opaque fragments can honestly offer. The artifact is the result pointer — Result.URI holds the uploads key and Result.Detail an ExportSummary.

What is left of this package's own state machine 

```mermaid
stateDiagram-v2
    [*] --> in_progress: Submit(export)
    [*] --> in_progress: Submit(erasure), no confirmation window
    [*] --> awaiting_confirmation: Submit(erasure), window > 0

    awaiting_confirmation --> in_progress: Confirm
    awaiting_confirmation --> cancelled: Cancel
    awaiting_confirmation --> cancelled: window lapses

    in_progress --> completed: fulfilled
    in_progress --> failed: the operation gave up
    in_progress --> cancelled: stopped mid-flight

    completed --> expired: artifact deleted
```

There is no cycle in it any more, and that is the shape of what the port removed. pending → processing → pending was a retry, and retries are the operation's now, along with the lease, the attempt counter, the backoff schedule, and the poll loop that drove all three. What is left are the states that were never the operation's to hold:

  • awaiting_confirmation, because a request nobody has confirmed has no operation at all. Folding it in would have meant an operation that exists in order to not run, which is a queue entry pretending to be a consent record.
  • cancelled by a lapsed window, for the same reason.
  • expired, because the artifact outlives the operation that produced it and is swept on its own schedule. An operation is done when the export is delivered; the file in the bucket is a separate obligation with a separate clock.

And three that look like the operation's and are not. in_progress is coarser than pending-or-running on purpose — this row does not know whether a worker has picked the work up, and Request.OperationID points at the thing that does. completed is written by the runner in the same transaction as the artifact reference and the audit entry, because those three facts have to commit together. failed is written on the operation's final attempt, which is the only moment at which "nobody is getting an answer" is a true thing to record; see operations.Attempt, which exists because of this.

The retention windows differ, and that is the practical reason for two records rather than one. An operation is reaped in weeks; this row is kept for years, because it is the statutory record that a named person asked and when.

Two-phase confirmation is opt-in: a ServiceConfig.ConfirmationWindow of zero — the default — starts an erasure on submission and Confirm is never needed. Turning it on is the difference between an accidental erasure being a support ticket and being unrecoverable, and regulation generally permits a verification step.

Assembly

The fulfiller supplies the runners and registers them; an operations Worker over the same registry runs them. 

fulfiller, err := dataprivacy.NewFulfiller(ctx, &dataprivacy.FulfillerConfig{}, store, registry,
    dataprivacy.WithFulfillerUploadManager(uploader),
    dataprivacy.WithFulfillerCompressor(compressor),
    dataprivacy.WithFulfillerAuditRecorder(recorder),
    dataprivacy.WithFulfillerNotifier(notifier),
    dataprivacy.WithFulfillerURLSigner(
        dataprivacy.NewArtifactURLSigner(uploader, 15*time.Minute, false),
    ),
)
if err != nil {
    return err
}

if err = fulfiller.Register(operationsRegistry); err != nil {
    return err
}

svc, err := dataprivacy.NewService(ctx, &dataprivacy.ServiceConfig{}, store, operationsService)

Every process that submits registers the kinds too, not only the ones that run them: operations resolves a kind at Start so that an unrunnable operation is refused at submission rather than discovered in a worker an hour later. In practice the API process builds a Fulfiller as well and simply never runs a worker.

The signer is what puts a working link in the completion mail. Without it the notification still goes out, saying the export is ready and to sign in for it — which is the right message when a link cannot be handed out, and the wrong one when it merely was not wired.

The Sweeper belongs on the jobs scheduler rather than on a ticker of its own, so the sweep runs once across a fleet: 

sweeper, err := dataprivacy.NewSweeper(ctx, &dataprivacy.SweeperConfig{}, store,
    dataprivacy.WithSweeperUploadManager(uploader),
)
if err != nil {
    return err
}

if err = scheduler.Register(sweeper.Job(jobs.MustCron("0 * * * *"), 30*time.Minute)); err != nil {
    return err
}

A deployment that runs the operations worker and not the Sweeper accumulates artifacts forever. That is the failure this package's design is most anxious about, which is why the sweep is a named, schedulable thing rather than a flag. Schedule operations.Service.Recover beside it for the reason operations gives: without it, a request whose enqueue was lost waits for nothing.

One sizing constraint comes with the port, and it is the kind that presents as a hang rather than an error. A runner holds a database transaction for the whole of its work — every eraser shares one, and an export's completion is one — while the operation's progress flush writes to the operations table beside it. Both draw from the same connection pool, so a pool without spare capacity deadlocks: size it for the operations worker's concurrency plus one connection per running operation, not for its concurrency alone.

Asking after a request

There are two reads and they answer different questions.

Service.Get returns the request: what was asked, by whom, when it is due, whether the artifact still exists. operations/http, mounted against Request.OperationID, returns how it is going — the state, the domains completed, the bytes collected, the structured error, and a server-sent event stream of the same. There is no status endpoint in this package, because there is no version of one that would be better than the one every other long-running thing in the application already has.

The operation's owner is the subject's ID, which is what lets that endpoint scope a read to the person it is about.

Delivering the artifact

The artifact is canonical JSON, then compressed, then optionally encrypted. Two delivery paths exist and they are not interchangeable:

  • Service.Download mints an expiring signed URL straight to storage. The bytes never pass through the application.
  • Service.Open streams the artifact, reversing compression and encryption. It works with every provider, at the cost of proxying.

Configuring an encryptor disables Download, and that is enforced rather than documented — see ErrArtifactEncrypted. A signed URL hands the client the stored object, and the stored object under encryption is base64 ciphertext. A subject who followed that link would get a file they cannot open, and would find out some days into a statutory window.

Encryption and the audit log

Two things this package touches are cryptographically load-bearing, and both are worth knowing about before wiring it up.

An artifact encrypted at rest is only as recoverable as the key. Losing the key turns every unexpired artifact into garbage, and the subjects waiting on them have a deadline. Encryption is therefore off unless configured.

Erasure and backups

Deleting a row erases it from the live database and from nowhere else. With any real retention window, an erasure completed today leaves the subject present in every snapshot taken before it — for the length of that window — and no amount of DELETE reaches a snapshot, because the media is not writable.

WithFulfillerShredder closes that, by destroying the subject's data key rather than only their rows: every column encrypted under that key becomes noise at once, in the live database and in every backup that already shipped. See cryptography/shredding, and read what it says about where the keys table lives before wiring it, because a keys table backed up alongside the data it protects hands everything back on the first restore.

The shred runs before the erasers and outside their transaction. Both are deliberate and both are explained at Fulfiller.erase. A scoped request does not shred at all — a data key spans every scope its subject appears in — and says so in Request.Retained rather than quietly doing less than it was asked.

The audit log is a hash chain, which means audit entries about a subject cannot simply be deleted or anonymized — either would make audit.Reader.Verify report tampering, for the rest of that scope's history. dataprivacy/auditerasure exists to do the part that is sound: delete whole audit scopes belonging to the subject, and report the rest as retained with a stated basis. It is registered explicitly, and an operator who wants the audit log left entirely alone simply does not register it.

What is recorded

Every submission, confirmation, cancellation, completion, and artifact access is written to the audit log when a Recorder is configured, in the same transaction as the state change it describes. "Who exported this person's data" is itself sensitive, and a system that can produce an export without leaving a record of who asked has a data exfiltration path with no alarm on it.

The audit entries carry the subject's ID and nothing else about them. An audit log is durable by design, and copying a person's data into the log that records the request to export it would defeat both. The operation carries even less: its request is a Job holding a request ID, and the runner reads the rest from the row.

Deadlines

Request.DueAt is stamped at submission from the configured response window — thirty days by default, GDPR's figure rather than CCPA's forty-five, because a deadline that is too early produces a gauge somebody looks at and one that is too late produces a fine. The Sweeper samples dataprivacy_requests_overdue from it.

Alerting on that gauge is left to the operator. The number is a fact; what counts as an incident is a policy, and this package has no business holding an opinion about which of a consumer's jurisdictions applies.

Upgrading

There is no migration from a dataprivacy_requests row that was mid-processing under the old worker to an operation, and none is offered. The old row carries an attempt count and a lease against a state machine that no longer exists, and inventing an operation for it would either re-run work that was half done or record a completion nothing performed. A subject with a statutory deadline is the wrong person to be approximately right about.

So drain before deploying: stop accepting requests, let the old worker finish what it claimed, and deploy once the table holds nothing in pending or processing. A deployment that cannot drain can run both releases for one release's overlap, with the old worker fulfilling the old rows while the new one starts operations for everything submitted after.

The schema changes with it. next_attempt, claimed_until, and attempts are gone, operation_id is new, and the pending and processing statuses are replaced by in_progress. Render the DDL from dataprivacy/migrations as usual; a table that already exists needs the columns added and dropped by hand, because this package ships no numbered migrations — see that package for why.

Index

Examples

Constants

View Source 
const (
	// DefaultResponseWindow is how long a request may take before it is
	// overdue.
	//
	// Thirty days is GDPR's window; CCPA allows forty-five. The stricter of the
	// two is the default because a deadline that is too early produces a gauge
	// somebody looks at, and one that is too late produces a fine.
	DefaultResponseWindow = 30 * 24 * time.Hour

	// DefaultArtifactTTL is how long an export artifact survives before the
	// sweeper deletes it.
	//
	// Seven days. The artifact contains everything an application knows about a
	// person, and the single worst outcome available to this package is leaving
	// one in a bucket indefinitely. Long enough that somebody on holiday can
	// still fetch it; short enough that it is not a permanent object.
	DefaultArtifactTTL = 7 * 24 * time.Hour

	// DefaultSignedURLTTL is how long a download URL is valid. Minutes, not
	// days: the link is mailed to the subject and mail is not a confidential
	// channel, so the window in which an intercepted link is useful should be
	// the window in which somebody clicks it.
	DefaultSignedURLTTL = 15 * time.Minute

	// DefaultArtifactPathPrefix is the storage prefix artifacts are written
	// under.
	DefaultArtifactPathPrefix = "dataprivacy/exports"

	// DefaultCollectorConcurrency is how many of one request's collectors run at
	// once.
	DefaultCollectorConcurrency = 4

	// DefaultCollectorTimeout bounds one collector. It exists so that one slow
	// domain costs its own section rather than the whole export — which is the
	// entire reason collection is per-key.
	DefaultCollectorTimeout = 30 * time.Second

	// DefaultFulfillmentTimeout bounds one whole attempt at one request.
	DefaultFulfillmentTimeout = 10 * time.Minute

	// DefaultMaxAttempts is how many times an operation fulfilling a privacy
	// request may be claimed before it is failed.
	//
	// Three, which is lower than the operations worker's own default of five,
	// and lower on purpose. One attempt here is a fan-out over every registered
	// domain against the application's own database, so the attempts are
	// expensive; and a request that is going to fail is worth failing early,
	// while there is still time inside the statutory window for somebody to fix
	// the cause and for the subject to be served.
	DefaultMaxAttempts = 3

	// DefaultMaxDocumentBytes caps the assembled export before it is written.
	//
	// A collector that answers a bad subject ID with its entire table is a bug
	// that presents as an out-of-memory kill in the worker, taking every other
	// in-flight operation with it. Failing the one request loudly is better.
	DefaultMaxDocumentBytes int64 = 512 << 20 // 512 MiB

	// DefaultSweepInterval is the recommended cadence for running the Sweeper.
	//
	// It is a suggestion for the caller's scheduler, not something this package
	// acts on: the Sweeper has no ticker of its own and does one pass per call.
	// It was previously also a config field, which read as though setting it made
	// the Sweeper run on that interval — nothing ever did.
	DefaultSweepInterval = time.Hour

	// DefaultSweepBatchSize caps how much one sweep tick does.
	DefaultSweepBatchSize = 100

	// DefaultRequestRetention is how long a terminal request record is kept
	// before the sweeper reaps it.
	//
	// A record of a privacy request is itself personal data — it says that a
	// named person asked, and when — so keeping it forever is the mistake this
	// package would otherwise make on every consumer's behalf. Three years
	// outlasts any plausible dispute about whether a request was honored while
	// not being indefinite.
	DefaultRequestRetention = 3 * 365 * 24 * time.Hour
)
View Source 
const (
	// KindExport is the operations kind that collects, packages, and delivers a
	// subject access request.
	KindExport = "dataprivacy.export"

	// KindErasure is the operations kind that shreds and erases.
	KindErasure = "dataprivacy.erasure"
)

The operation kinds this package registers. They are written into every operation row this package starts, so they are stable by contract: renaming one strands every operation already queued under the old name, which the operations worker fails with operations.CodeUnknownKind.

View Source 
const (
	// CodeRequestGone is the code for a runner whose request row is not there.
	// It means retention reaped it, or it never existed — an operation started
	// against an ID nothing wrote.
	CodeRequestGone = "dataprivacy_request_gone"

	// CodeNotInProgress is the code for a runner whose request row moved on
	// while the operation was queued: cancelled, or already fulfilled.
	CodeNotInProgress = "dataprivacy_request_not_in_progress"

	// CodeCancelled is the code a runner that stopped because it was asked to
	// records. The operation is recorded as cancelled whatever the runner
	// returns, so this is what an operator reads to see it exited at a point it
	// could describe rather than mid-write.
	CodeCancelled = "dataprivacy_cancelled"

	// CodeEverySectionFailed is the code for an export in which no collector
	// succeeded. See ErrEverySectionFailed.
	CodeEverySectionFailed = "dataprivacy_every_section_failed"

	// CodeDocumentTooLarge is the code for an assembled export past
	// FulfillerConfig.MaxDocumentBytes.
	CodeDocumentTooLarge = "dataprivacy_document_too_large"

	// CodeNoErasers is the code for an erasure with nothing registered to run.
	CodeNoErasers = "dataprivacy_no_erasers"

	// CodeUnknownRequestType is the code for a request row naming a type this
	// build does not implement.
	CodeUnknownRequestType = "dataprivacy_unknown_request_type"
)

Failure codes this package attaches to the operations it fails, for a client that has to branch on why rather than read a sentence.

View Source 
const DefaultSweepJobName = "dataprivacy-sweep"

DefaultSweepJobName is the name the Sweeper's jobs.Job carries.

It is a constant because a job's name is its lock key: two replicas that disagree about it both run the sweep, and the sweep deletes things.

View Source 
const DefaultTablePrefix = ""

DefaultTablePrefix is the namespace the dataprivacy tables carry when none is configured, which is none — rendering dataprivacy_requests.

The dataprivacy_ segment is the schema's, not the caller's: a table always says which package created it. Setting a namespace of "ddb" renders ddb_dataprivacy_requests, for a database shared between applications. A namespace must not end in '_'; database/ddl supplies the separator.

View Source 
const DocumentFormat = "dataprivacy.export.v1"

DocumentFormat tags the artifact's framing. It is the first thing a reader should look at, and it exists so that a v2 layout is distinguishable from a v1 one by something better than guessing from which keys are present.

Variables

View Source 
var (
	// ErrNilStore indicates a nil Store. It wraps errors.ErrNilInputParameter,
	// so a caller may check either.
	ErrNilStore = platformerrors.Wrap(platformerrors.ErrNilInputParameter, "nil dataprivacy store")

	// ErrNilDatabaseClient indicates a nil database.Client. It wraps
	// errors.ErrNilInputParameter, so a caller may check either.
	ErrNilDatabaseClient = platformerrors.Wrap(platformerrors.ErrNilInputParameter, "nil database client")

	// ErrNilExecutor indicates a Store method that runs in the caller's
	// transaction was called without one.
	ErrNilExecutor = platformerrors.Wrap(platformerrors.ErrNilInputParameter, "nil query executor")

	// ErrNilRequest indicates a nil *Request.
	ErrNilRequest = platformerrors.Wrap(platformerrors.ErrNilInputParameter, "nil dataprivacy request")

	// ErrNilOperations indicates a Service built without an operations.Service.
	//
	// It has no default. A Service that could not start operations would record
	// requests nothing ever fulfills, which looks exactly like a working Service
	// until a subject's statutory window runs out.
	ErrNilOperations = platformerrors.Wrap(platformerrors.ErrNilInputParameter, "nil dataprivacy operations service")

	// ErrNotInProgress indicates a runner whose request row is not in
	// StatusInProgress.
	//
	// It means the request left the state the runner was started for while the
	// operation was queued or running: cancelled, already completed by an
	// earlier attempt, or reaped. It is unretryable — none of those become
	// StatusInProgress again by waiting.
	ErrNotInProgress = platformerrors.New("dataprivacy request is not in progress")

	// ErrEmptySubjectID indicates a Subject with no ID. Every request is about
	// somebody, and a request about nobody would fan out over every collector
	// asking for the empty string's data — which some of them will answer.
	ErrEmptySubjectID = platformerrors.New("empty dataprivacy subject ID")

	// ErrUnknownRequestType indicates a RequestType outside the two this package
	// implements.
	ErrUnknownRequestType = platformerrors.New("unknown dataprivacy request type")

	// ErrRequestNotFound indicates a request ID that is not in the table. It may
	// mean the request never existed, or that retention has swept it.
	ErrRequestNotFound = platformerrors.New("dataprivacy request not found")

	// ErrNotAwaitingConfirmation indicates a Confirm or Cancel naming a request
	// that is not waiting for one — because it was never two-phase, because it
	// has already been confirmed, or because its confirmation window lapsed.
	ErrNotAwaitingConfirmation = platformerrors.New("dataprivacy request is not awaiting confirmation")

	// ErrNoCollectors indicates an export Service built with no registered
	// Collector. It is refused at construction rather than at fulfillment: an
	// export service with no collectors produces a valid, empty, and entirely
	// wrong artifact, and a subject who receives one has been told that nothing
	// is held about them.
	ErrNoCollectors = platformerrors.New("no dataprivacy collectors registered")

	// ErrNoErasers indicates an erasure Service built with no registered Eraser,
	// refused for the same reason as ErrNoCollectors and with a worse failure
	// mode: an erasure that erases nothing reports success.
	ErrNoErasers = platformerrors.New("no dataprivacy erasers registered")

	// ErrDuplicateKey indicates two registrations under one key. Keys become
	// section names in the artifact, so a silent overwrite would drop a domain
	// from every export without any signal that it had.
	ErrDuplicateKey = platformerrors.New("duplicate dataprivacy registration key")

	// ErrInvalidKey indicates a registration key that is empty or is not a
	// plain identifier. Keys are JSON object keys in the artifact and path
	// segments in telemetry, so they are restricted rather than escaped.
	ErrInvalidKey = platformerrors.New("invalid dataprivacy registration key")

	// ErrArtifactUnavailable indicates a Download or Open for a request that has
	// no artifact: an erasure, a request that has not completed, or one whose
	// artifact has expired and been deleted.
	ErrArtifactUnavailable = platformerrors.New("dataprivacy artifact is unavailable")

	// ErrArtifactEncrypted indicates a Download against a Service configured
	// with an Encryptor.
	//
	// The two are genuinely incompatible rather than merely awkward. A signed
	// URL hands the client the stored object, and the stored object is
	// ciphertext this package base64s — a subject who follows that link gets a
	// file they cannot open, and finds out thirty days into a statutory window.
	// Encryption at rest and direct-to-bucket delivery are a choice between two
	// things, so configuring both fails here rather than at the subject.
	ErrArtifactEncrypted = platformerrors.New("dataprivacy artifact is encrypted and cannot be delivered by signed URL")

	// ErrNoURLSigner indicates a Download against an UploadManager that cannot
	// sign URLs. Not every provider can — the filesystem one certainly cannot —
	// and Open is the path that works everywhere.
	ErrNoURLSigner = platformerrors.New("dataprivacy upload manager cannot sign URLs")
)
View Source 
var (
	// ErrNoUploadManager indicates an export Fulfiller with nowhere to write. It
	// is refused at construction: a fulfiller that collects eleven domains and
	// then discovers it has no storage has already done all the expensive work
	// and still fails.
	ErrNoUploadManager = platformerrors.New("no dataprivacy upload manager configured")

	// ErrDocumentTooLarge indicates an assembled export past MaxDocumentBytes.
	ErrDocumentTooLarge = platformerrors.New("dataprivacy export document exceeds configured maximum")

	// ErrEverySectionFailed indicates an export in which no collector succeeded.
	//
	// A partial export is delivered with a manifest naming the gaps, because
	// most of somebody's data plus an honest account of the rest is worth
	// having. An export with no data at all is not a partial export — it is a
	// file asserting that nothing is held about a person, which is the one
	// wrong answer this package could give.
	ErrEverySectionFailed = platformerrors.New("no dataprivacy collector succeeded")

	// ErrInvalidFragment indicates a Collector that returned something that is
	// not valid JSON. It is caught before assembly rather than at read time,
	// because a malformed fragment would otherwise produce an artifact that
	// cannot be parsed at all — turning one domain's bug into a total loss.
	ErrInvalidFragment = platformerrors.New("dataprivacy collector returned invalid JSON")

	// ErrCollectorPanicked indicates a Collector that panicked. It is that
	// section's failure rather than the operation's: a nil map access in one
	// domain should cost that domain's section, not the export.
	ErrCollectorPanicked = platformerrors.New("dataprivacy collector panicked")

	// ErrEraserPanicked indicates an Eraser that panicked. Unlike a collector
	// panic this aborts the whole erasure, because every eraser shares one
	// transaction and a panic mid-way through leaves no coherent partial state
	// to record.
	ErrEraserPanicked = platformerrors.New("dataprivacy eraser panicked")
)

Functions

func KindFor 

func KindFor(t RequestType) (string, bool)

KindFor names the operation kind that fulfills a request type, and reports whether there is one.

func NewArtifactURLSigner 

func NewArtifactURLSigner(
	manager uploads.UploadManager,
	ttl time.Duration,
	encrypted bool,
	opts ...URLSignerOption,
) func(ctx context.Context, req *Request) (string, time.Time)

NewArtifactURLSigner builds the signer a Fulfiller hands to WithFulfillerURLSigner, so a completion notification can carry a working download link.

It exists because the Fulfiller cannot hold a Service — a Service is the thing that reads what the Fulfiller writes — but the notification is only useful with a URL in it. The manager and TTL must be the ones the Service would use, or the subject gets a link into the wrong bucket.

It declines to sign in exactly the cases Service.Download refuses: an encrypted artifact, and a provider that cannot sign. An empty URL is not an error here — the notification simply tells the subject their export is ready and to sign in for it, which is the correct message when a link cannot be handed out.

Types

type ActorResolver 

type ActorResolver func(ctx context.Context) audit.Actor

ActorResolver names the principal responsible for an action, for the audit entry this package writes.

It reads from the context because Submit's signature belongs to the subject, not to whoever is acting on their behalf — and those differ in exactly the case that matters. A support agent running an export for a customer is the event worth recording, and "who exported this person's data" is not answerable from the Subject alone.

Without one, actions are attributed to audit.ActorSystem, which is honest for a self-service portal and misleading for a staff tool.

type Collector 

type Collector interface {
	Collect(ctx context.Context, subject Subject) (json.RawMessage, error)
}

Collector produces one domain's view of a subject.

Collect returns already-encoded JSON rather than a value to be marshaled, and that is the load-bearing decision in this package. The prior art this generalizes had every domain mutate one shared aggregate struct, so adding a domain meant editing a central type that imported every domain package — a cost paid on every schema change, by the one file most likely to conflict. A library cannot have that type at all, and it turns out not to need one: an opaque fragment per key composes into a document without the library knowing what any of it means.

Returning nil, nil is how a domain says "nothing about this subject". The section is then omitted from the artifact rather than written as null, so an export's sections are the domains that actually held something.

A Collector must not return partially-collected data alongside an error. The fragment is used or the error is recorded; there is no path that writes both.

There is deliberately no as-of time in this signature, and it is worth being clear about what that means. A fragment is the domain's state at the instant Collect ran, which is when a worker got to the operation — not when the subject asked. The two differ by the queue depth plus any retries, and because collectors run concurrently they differ from each other as well: an artifact is a smear across the collection window rather than a snapshot at any one instant. Manifest.GeneratedAt is the only time the artifact states.

This matches the ordinary reading of a subject access request — the data held when the response is produced — and it is the only thing a library can promise. Bounding an export to data created on or before Request.RequestedAt would have to be a parameter here, honored by every registered Collector, and nothing in this package could enforce it: a domain with no reliable creation timestamp cannot answer the question at all, and one that ignored the bound would be silently wrong in the direction that matters. An application whose jurisdiction or dispute posture needs that guarantee has to implement it in its collectors, and know that it has.

Example

A Collector returns one domain's view of a subject as already-encoded JSON. The library never looks inside it, which is what lets a domain be added by registration rather than by editing a shared type. 

package main

import (
	"context"
	"encoding/json"
	"fmt"

	"github.com/primandproper/platform-go/v12/dataprivacy"
)

func main() {
	identity := dataprivacy.CollectorFunc(func(_ context.Context, subject dataprivacy.Subject) (json.RawMessage, error) {
		// In a real collector this is a query against the domain's own tables.
		return json.Marshal(map[string]string{
			"id":    subject.ID,
			"email": "someone@example.com",
		})
	})

	fragment, err := identity.Collect(context.Background(), dataprivacy.Subject{ID: "user-1"})
	if err != nil {
		panic(err)
	}

	fmt.Println(string(fragment))
}

Output:
{"email":"someone@example.com","id":"user-1"}

type CollectorFunc 

type CollectorFunc func(ctx context.Context, subject Subject) (json.RawMessage, error)

CollectorFunc adapts a function to Collector.

func (CollectorFunc) Collect 

func (f CollectorFunc) Collect(ctx context.Context, subject Subject) (json.RawMessage, error)

Collect implements Collector.

type Document 

type Document struct {
	// Data maps a section name to that domain's fragment, verbatim as its
	// Collector returned it.
	Data map[string]json.RawMessage `json:"data"`
	// Manifest describes the document.
	Manifest Manifest `json:"manifest"`
}

Document is the artifact's top-level shape: what this is, and the data.

The two are siblings rather than the manifest being folded into the data, so that a section named "manifest" — which a domain is entitled to register — cannot collide with the framing.

func (*Document) Complete 

func (d *Document) Complete() bool

Complete reports whether every registered section was collected.

type EmailNotifier 

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

EmailNotifier sends completion mail through an email.Emailer.

It is deliberately plain. The message it renders is a serviceable default rather than a good one, and an application with a template system should implement Notifier directly instead — this exists so that wiring up the package end to end does not require writing one first.

func NewEmailNotifier 

func NewEmailNotifier(
	emailer email.Emailer,
	from Recipient,
	resolve RecipientResolver,
	opts ...EmailNotifierOption,
) (*EmailNotifier, error)

NewEmailNotifier builds a Notifier over an email.Emailer.

from is the sender. resolve turns a Subject into an address; it is required, and see RecipientResolver for why the library cannot supply one.

func (*EmailNotifier) Notify 

func (n *EmailNotifier) Notify(ctx context.Context, notification *Notification) error

Notify implements Notifier.

type EmailNotifierOption 

type EmailNotifierOption func(*EmailNotifier)

EmailNotifierOption configures an EmailNotifier.

func WithMessageRenderer 

func WithMessageRenderer(renderer MessageRenderer) EmailNotifierOption

WithMessageRenderer replaces the default message.

type Eraser 

type Eraser interface {
	Erase(ctx context.Context, q database.SQLQueryExecutor, subject Subject) (ErasureOutcome, error)
}

Eraser removes or anonymizes one domain's data about a subject.

It is deliberately separate from Collector rather than derived from it. Erasure is not the inverse of export: some data must be retained (financial records under tax law, audit entries under legitimate interest) and some must be anonymized in place rather than deleted, because a foreign key still points at it. Only the domain knows which of the three applies to each of its tables, and a library that inferred "erase everything you would have exported" would be confidently wrong about all of it.

Erase runs inside the request's transaction and must use the executor it is given rather than a handle of its own. Every registered eraser for one request shares that transaction, so an erasure is all-or-nothing: a subject is not left half-deleted across eleven domains because the ninth timed out.

Example

An Eraser reports what it destroyed, what it anonymized, and what it kept. Erasure is not the inverse of export: only the domain knows which of its tables must be retained and on what basis. 

package main

import (
	"context"
	"fmt"

	"github.com/primandproper/platform-go/v12/database"
	"github.com/primandproper/platform-go/v12/dataprivacy"
)

func main() {
	billing := dataprivacy.EraserFunc(func(
		_ context.Context,
		_ database.SQLQueryExecutor,
		_ dataprivacy.Subject,
	) (dataprivacy.ErasureOutcome, error) {
		return dataprivacy.ErasureOutcome{
			Deleted:    12,
			Anonymized: 3,
			Retained: map[string]string{
				"invoices": "financial records, retained 7 years under tax law",
			},
		}, nil
	})

	outcome, err := billing.Erase(context.Background(), nil, dataprivacy.Subject{ID: "user-1"})
	if err != nil {
		panic(err)
	}

	fmt.Println(outcome.Deleted, outcome.Anonymized)
	fmt.Println(outcome.Retained["invoices"])
}

Output:
12 3
financial records, retained 7 years under tax law

type EraserFunc 

type EraserFunc func(ctx context.Context, q database.SQLQueryExecutor, subject Subject) (ErasureOutcome, error)

EraserFunc adapts a function to Eraser.

func (EraserFunc) Erase 

func (f EraserFunc) Erase(ctx context.Context, q database.SQLQueryExecutor, subject Subject) (ErasureOutcome, error)

Erase implements Eraser.

type ErasureOutcome 

type ErasureOutcome struct {
	// Retained names what was kept and the legal basis for keeping it — the
	// string goes into the request record and, in practice, in front of a
	// regulator. "invoices: financial records, retained 7 years under
	// [statute]" is the shape that answers the question; "some data" is not.
	//
	// It is keyed so one domain can retain several things for different
	// reasons, which is the normal case rather than the exotic one.
	Retained map[string]string `json:"retained,omitempty"`
	// Deleted is how many rows were destroyed.
	Deleted int64 `json:"deleted"`
	// Anonymized is how many rows were kept but stripped of anything
	// identifying. A row that was both is counted once, here.
	Anonymized int64 `json:"anonymized"`
}

ErasureOutcome is what one domain did.

type ErasureSummary 

type ErasureSummary struct {
	// KeyShreddedAt is when the subject's data key was destroyed, or nil when no
	// shredder was configured or the request was scoped. Retained says which.
	KeyShreddedAt *time.Time `json:"keyShreddedAt,omitempty"`

	// Retained records, per eraser key, what was kept and why.
	Retained map[string]string `json:"retained,omitempty"`

	// Deleted and Anonymized are the totals summed across every eraser.
	Deleted    int64 `json:"deleted"`
	Anonymized int64 `json:"anonymized"`
}

ErasureSummary is what a completed erasure records in operations.Result.Detail. There is no Result.URI: an erasure produces no artifact, which is the whole of what distinguishes it from an export here.

type ExportSummary 

type ExportSummary struct {
	// GeneratedAt is when the artifact was assembled.
	GeneratedAt time.Time `json:"generatedAt"`

	// Failures maps a section name to why it is missing. Absent when the export
	// was complete, present — and the reason this type is worth having — when it
	// is a partial export that was delivered anyway.
	Failures map[string]string `json:"failures,omitempty"`

	// Bytes is the stored size of the artifact, after compression and
	// encryption.
	Bytes int64 `json:"bytes"`
}

ExportSummary is what a completed export records in operations.Result.Detail, beside the artifact's key in Result.URI.

It is deliberately narrower than Manifest, in two ways. It omits the subject, because the manifest travels inside the artifact — which is delivered to that person — and this travels in the operations table, read by a status endpoint that already knows who is asking. And it omits the section list, because the artifact's own manifest is the authority on what is in it.

What is left is exactly what the request row holds, and that is the property worth having: a duplicate execution that finds the export already recorded can rebuild this summary from the row rather than reporting a thinner one than the attempt that did the work.

type Fulfiller 

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

Fulfiller does the work behind this package's two operation kinds.

It is not a loop and owns no goroutine. That is the whole of what the port onto operations changed here: an operations Worker claims the work, leases it, retries it, and reports where it got to, and this supplies the two functions it calls. What used to be a poll interval, a batch size, a lease, an attempt counter, and a backoff schedule in this package is now one worker's, shared with every other long-running thing in the application.

Register it into an operations.Registry and run an operations.Worker over the same registry: 

f, err := dataprivacy.NewFulfiller(ctx, &dataprivacy.FulfillerConfig{}, store, registry,
    dataprivacy.WithFulfillerUploadManager(uploader),
)
// ...
if err = f.Register(operationsRegistry); err != nil {
    return err
}

func NewFulfiller 

func NewFulfiller(
	ctx context.Context,
	cfg *FulfillerConfig,
	store Store,
	registry *Registry,
	opts ...FulfillerOption,
) (*Fulfiller, error)

NewFulfiller builds a Fulfiller. It does not run anything; see Register.

ctx is used to validate the config and is not retained.

A registry with no collectors and no erasers is refused. So is an export capability with no storage. Both would produce a fulfiller that accepts work, does nothing useful, and reports success — and an erasure service that erases nothing while reporting success is the worst failure available here, because nobody goes looking for it.

func (*Fulfiller) Register 

func (f *Fulfiller) Register(registry *operations.Registry) error

Register adds this package's kinds to an operations registry: KindExport when there are collectors, KindErasure when there are erasers.

Every process that starts one of these operations has to register it too, not only the processes that run them — operations.Service.Start resolves the kind through the registry so that an unrunnable operation is refused at submission rather than discovered in a worker an hour later. In practice that means the API process builds a Fulfiller as well, with the same registry and the same storage, and simply never runs an operations.Worker.

Example

Both halves of the package run as operations. The Fulfiller supplies the two runners and registers them; an operations Worker over the same registry is what claims, leases, retries, and reports on them. 

package main

import (
	"context"
	"encoding/json"
	"fmt"

	"github.com/primandproper/platform-go/v12/database"
	"github.com/primandproper/platform-go/v12/dataprivacy"

	dataprivacymock "github.com/primandproper/platform-go/v12/dataprivacy/mock"
	"github.com/primandproper/platform-go/v12/operations"

	uploadsnoop "github.com/primandproper/platform-go/v12/uploads/noop"
)

func main() {
	domains := dataprivacy.NewRegistry()

	if err := domains.RegisterCollector("identity", dataprivacy.CollectorFunc(
		func(context.Context, dataprivacy.Subject) (json.RawMessage, error) {
			return json.RawMessage(`{}`), nil
		},
	)); err != nil {
		panic(err)
	}

	if err := domains.RegisterEraser("identity", dataprivacy.EraserFunc(
		func(context.Context, database.SQLQueryExecutor, dataprivacy.Subject) (dataprivacy.ErasureOutcome, error) {
			return dataprivacy.ErasureOutcome{}, nil
		},
	)); err != nil {
		panic(err)
	}

	// In a real assembly the store is a dataprivacy.NewSQLStore and the uploader
	// is real storage; the registration below is the whole of the wiring this
	// example is about.
	fulfiller, err := dataprivacy.NewFulfiller(
		context.Background(), &dataprivacy.FulfillerConfig{}, &dataprivacymock.StoreMock{}, domains,
		dataprivacy.WithFulfillerUploadManager(uploadsnoop.NewUploadManager()),
	)
	if err != nil {
		panic(err)
	}

	kinds := operations.NewRegistry()
	if err = fulfiller.Register(kinds); err != nil {
		panic(err)
	}

	// A process that only submits registers these too: operations resolves a
	// kind at Start, so an unrunnable operation is refused there rather than
	// discovered in a worker an hour later.
	fmt.Println(kinds.Kinds())
}

Output:
[dataprivacy.erasure dataprivacy.export]

type FulfillerConfig 

type FulfillerConfig struct {
	// ArtifactPathPrefix is the storage prefix artifacts are written under.
	// Defaults to DefaultArtifactPathPrefix.
	ArtifactPathPrefix string `env:"ARTIFACT_PATH_PREFIX" json:"artifactPathPrefix,omitempty" yaml:"artifactPathPrefix,omitempty"`

	// FulfillmentTimeout bounds one whole attempt at one request.
	//
	// It matters more than it looks now that the operation's lease is extended
	// by every progress flush: the lease no longer bounds anything, so this is
	// what stands between a wedged domain and an operation that never reaches a
	// terminal state.
	FulfillmentTimeout time.Duration `env:"FULFILLMENT_TIMEOUT" json:"fulfillmentTimeout,omitempty" yaml:"fulfillmentTimeout,omitempty"`

	// CollectorTimeout bounds one collector, so one slow domain costs its own
	// section rather than the export.
	CollectorTimeout time.Duration `env:"COLLECTOR_TIMEOUT" json:"collectorTimeout,omitempty" yaml:"collectorTimeout,omitempty"`

	// ArtifactTTL is how long an export artifact survives after completion,
	// stamped onto the request as ExpiresAt when the export is recorded.
	// Defaults to DefaultArtifactTTL.
	ArtifactTTL time.Duration `env:"ARTIFACT_TTL" json:"artifactTTL,omitempty" yaml:"artifactTTL,omitempty"`

	// MaxDocumentBytes caps the assembled export. Defaults to
	// DefaultMaxDocumentBytes.
	MaxDocumentBytes int64 `env:"MAX_DOCUMENT_BYTES" json:"maxDocumentBytes,omitempty" yaml:"maxDocumentBytes,omitempty"`

	// MaxAttempts is how many times an operation of either kind may be claimed
	// before it is failed. It becomes operations.Definition.MaxAttempts, and
	// zero means the operations worker's own ceiling.
	//
	// It is set here rather than left to that ceiling because a privacy request
	// is not a webhook replay: one attempt is a fan-out over every registered
	// domain, and the default is deliberately low so that a request which is
	// going to fail says so within the statutory window rather than at the end
	// of it. See DefaultMaxAttempts.
	MaxAttempts int `env:"MAX_ATTEMPTS" json:"maxAttempts,omitempty" yaml:"maxAttempts,omitempty"`

	// CollectorConcurrency is how many of one request's collectors run at once.
	//
	// It is bounded rather than unlimited because every collector queries the
	// application's own database, and a subject present in forty domains would
	// otherwise open forty concurrent queries on behalf of one background job.
	CollectorConcurrency int `env:"COLLECTOR_CONCURRENCY" json:"collectorConcurrency,omitempty" yaml:"collectorConcurrency,omitempty"`
}

FulfillerConfig configures the two operation runners.

It is most of what it used to be minus a whole category of knob. The poll interval, the batch size, the request concurrency, the lease, and the backoff schedule are the operations worker's now — one worker, one set of numbers, shared with every other long-running thing in the application — and this is left with the settings that are genuinely about privacy requests.

func (*FulfillerConfig) EnsureDefaults 

func (cfg *FulfillerConfig) EnsureDefaults()

EnsureDefaults fills unset knobs with the package defaults.

func (*FulfillerConfig) ValidateWithContext 

func (cfg *FulfillerConfig) ValidateWithContext(ctx context.Context) error

ValidateWithContext validates a FulfillerConfig.

type FulfillerOption 

type FulfillerOption func(*Fulfiller)

FulfillerOption configures a Fulfiller.

func WithFulfillerActorResolver 

func WithFulfillerActorResolver(resolver ActorResolver) FulfillerOption

WithFulfillerActorResolver supplies the principal recorded in audit entries.

func WithFulfillerAuditRecorder 

func WithFulfillerAuditRecorder(recorder audit.Recorder) FulfillerOption

WithFulfillerAuditRecorder attaches the audit log completions are recorded in.

The completion entry is the one that says what was actually disclosed or destroyed, and it is written in the same transaction as the state change it describes.

func WithFulfillerClock 

func WithFulfillerClock(c clock.Clock) FulfillerOption

WithFulfillerClock swaps the clock stamping completions, artifact expiry, and audit entries.

func WithFulfillerCompressor 

func WithFulfillerCompressor(compressor compression.Compressor) FulfillerOption

WithFulfillerCompressor compresses artifacts before they are stored.

Worth setting. An export is JSON assembled from every domain in an application, which is the most compressible shape there is — and the artifact is written once and read at most once, so the compression is nearly free.

func WithFulfillerEncryptor 

func WithFulfillerEncryptor(encryptor encryption.Encryptor) FulfillerOption

WithFulfillerEncryptor encrypts artifacts at rest.

It changes what delivery is possible: an encrypted artifact cannot be handed out as a signed URL, because the subject would receive ciphertext. See ErrArtifactEncrypted. Configure the Service with the matching decryptor.

func WithFulfillerLogger 

func WithFulfillerLogger(logger logging.Logger) FulfillerOption

WithFulfillerLogger attaches a logger. A failing collector is reported through it and nowhere else — there is no caller to return it to — so without one a domain that has been failing to collect for a week is visible only in metrics.

func WithFulfillerMetricsProvider 

func WithFulfillerMetricsProvider(metricsProvider metrics.Provider) FulfillerOption

WithFulfillerMetricsProvider attaches a metrics provider.

func WithFulfillerNotifier 

func WithFulfillerNotifier(notifier Notifier) FulfillerOption

WithFulfillerNotifier supplies who to tell when a request finishes.

func WithFulfillerShredder 

func WithFulfillerShredder(shredder shredding.Shredder) FulfillerOption

WithFulfillerShredder destroys the subject's data key as part of an erasure, so the erasure reaches media that deletion cannot.

Without it an erasure deletes rows, and the rows stay in every backup taken before it ran — for the whole retention window, which is the part of "we erased you" that is not true. Destroying the key makes every ciphertext it protected unreadable everywhere at once, including in snapshots nobody can write to.

It is not a substitute for the erasers. Only the columns an application chose to encrypt under the subject's key are covered, the shred does not run inside their transaction, and what it destroys it destroys whether or not they succeed. See Fulfiller.erase's ordering, which is deliberate and stated in the source.

Setting it on a Fulfiller whose application encrypts nothing per subject is harmless and close to pointless: every erasure writes a tombstone and destroys nothing, which Request.KeyShreddedAt will happily record.

func WithFulfillerTracerProvider 

func WithFulfillerTracerProvider(tracerProvider tracing.Provider) FulfillerOption

WithFulfillerTracerProvider attaches a tracer provider. The spans it produces hang under the operations worker's, so a slow export reads as one trace from the claim through to the domain that took the time.

func WithFulfillerURLSigner 

func WithFulfillerURLSigner(signer func(ctx context.Context, req *Request) (url string, expiresAt time.Time)) FulfillerOption

WithFulfillerURLSigner supplies how a notification's download URL is minted.

It exists so the Fulfiller can hand the subject a link without holding a Service — which would be circular, since a Service is the thing that reads what this Fulfiller writes. The signer returns the URL and its expiry; an empty URL means the notification carries no link, which is correct for encrypted artifacts and for providers that cannot sign.

func WithFulfillerUploadManager 

func WithFulfillerUploadManager(manager uploads.UploadManager) FulfillerOption

WithFulfillerUploadManager supplies the storage artifacts are written to. Required for exports; an erasure-only Fulfiller does not need it.

type Job 

type Job struct {
	// RequestID names the dataprivacy request this operation fulfills.
	RequestID string `json:"requestID"`
}

Job is the operation request both of this package's kinds are started with.

It carries the request ID and nothing else, and that is deliberate. The operation row stores its request encoded in a column, and everything else about a privacy request — who it is about, what scope it covers — is exactly the material this package works hardest to keep out of places it does not need to be. The runner reads the request row, which is the one place that data has to live.

It also means there is one source of truth for what the request says. A subject copied into the operation at submission and read back an hour later would be a second copy that nothing keeps current.

type Manifest 

type Manifest struct {
	// GeneratedAt is when the artifact was assembled.
	//
	// It is not Request.RequestedAt, and the gap between them is the queue wait
	// plus any retries. Sections are collected as of roughly this instant rather
	// than as of the request — see Collector — so this is the time the document
	// describes.
	GeneratedAt time.Time `json:"generatedAt"`
	// Failures maps a section name to why it is missing. Absent when the export
	// was complete.
	Failures map[string]string `json:"failures,omitempty"`
	// Format is DocumentFormat.
	Format string `json:"format"`
	// RequestID is the request this artifact answers.
	RequestID string `json:"requestID"`
	// Subject is who it is about.
	Subject Subject `json:"subject"`
	// Sections are the section names present in Data, sorted.
	Sections []string `json:"sections"`
}

Manifest describes what an artifact contains and, more importantly, what it does not.

Failures is the field that earns this type. An export assembled from eleven domains where one timed out is still worth delivering — the subject is entitled to the other ten, and the statutory clock does not stop while a flaky domain is fixed — but delivering it without saying so would be a document that quietly asserts the missing data does not exist. Naming the gap is the difference between a partial answer and a wrong one.

Example

A partial export is delivered with a manifest naming what is missing, rather than failing outright or silently omitting the gap. 

package main

import (
	"encoding/json"
	"fmt"

	"github.com/primandproper/platform-go/v12/dataprivacy"
)

func main() {
	doc := &dataprivacy.Document{
		Data: map[string]json.RawMessage{
			"identity": json.RawMessage(`{"email":"someone@example.com"}`),
		},
		Manifest: dataprivacy.Manifest{
			Format:    dataprivacy.DocumentFormat,
			RequestID: "req-1",
			Sections:  []string{"identity"},
			Failures:  map[string]string{"billing": "context deadline exceeded"},
		},
	}

	fmt.Println(doc.Complete())
	fmt.Println(doc.Manifest.Failures["billing"])
}

Output:
false
context deadline exceeded

type MessageRenderer 

type MessageRenderer func(notification *Notification, to Recipient) (subject, htmlBody string)

MessageRenderer builds the subject line and HTML body for a notification.

type Notification 

type Notification struct {
	// ExpiresAt is when DownloadURL stops working. Zero when there is no URL.
	ExpiresAt time.Time
	// Request is the request that reached a terminal state.
	Request *Request
	// DownloadURL is a freshly minted, expiring URL for the artifact. Empty for
	// an erasure, for a failure, and whenever the Service cannot mint one — an
	// encrypted artifact or a provider that cannot sign.
	//
	// It is minted at notification time rather than at completion so that its
	// short expiry starts when the subject is told, not when the runner
	// finished. A URL that expired before the mail was delivered is a support
	// ticket the subject opens on day 29.
	DownloadURL string
}

Notification is what a Notifier is handed when a request reaches a terminal state.

type Notifier 

type Notifier interface {
	Notify(ctx context.Context, notification *Notification) error
}

Notifier tells somebody a request is done.

It is an interface rather than a fixed email template because the library cannot write the message. It does not know the subject's email address — a Subject is an opaque ID — nor the tone, the language, or the legal boilerplate the jurisdiction requires. What it does know is when to send, and that is what this seam supplies.

A Notifier's error does not fail the request. The export was produced and the erasure ran; a mail server being down does not undo either, and retrying the fulfillment to retry the mail would re-run the collectors. Failures are logged and counted.

type NotifierFunc 

type NotifierFunc func(ctx context.Context, notification *Notification) error

NotifierFunc adapts a function to Notifier.

func (NotifierFunc) Notify 

func (f NotifierFunc) Notify(ctx context.Context, notification *Notification) error

Notify implements Notifier.

type Recipient 

type Recipient struct {
	// Address is the email address. Required.
	Address string
	// Name is the display name, if there is one.
	Name string
}

Recipient is who to mail, resolved from a Subject.

type RecipientResolver 

type RecipientResolver func(ctx context.Context, subject Subject) (*Recipient, error)

RecipientResolver maps a Subject to who should be told about it.

It is required by NewEmailNotifier and has no default, because the mapping is the one piece of this that only the application has: a Subject carries an opaque ID, and turning that into an address is a database read this package has no business performing. Returning a nil Recipient with a nil error means "do not mail anyone about this", which is the right answer for a subject who has just been erased.

type Registry 

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

Registry is the set of domains that know how to collect and erase.

Registration replaces the god-struct the prior art aggregated into. There, every domain wrote into one shared type, so adding a domain meant editing a central file that imported every domain package — a cost paid on every schema change by the one file most likely to conflict, and one that grew two fields in a single month. Here a domain announces itself and the library composes what it gets, so adding one touches one call site and nothing else.

A Registry is built during startup and read concurrently thereafter. It is not safe to register into one that a Fulfiller is already running against, and nothing here pretends otherwise: registration is a wiring-time activity, and a mutex would only make an ordering bug quieter.

Example

Adding a domain is a registration, not an edit to a central type that imports every domain package. 

package main

import (
	"context"
	"encoding/json"
	"fmt"

	"github.com/primandproper/platform-go/v12/dataprivacy"
)

func main() {
	registry := dataprivacy.NewRegistry()

	for _, key := range []string{"identity", "billing", "webhooks"} {
		if err := registry.RegisterCollector(key, dataprivacy.CollectorFunc(
			func(context.Context, dataprivacy.Subject) (json.RawMessage, error) {
				return json.RawMessage(`{}`), nil
			},
		)); err != nil {
			panic(err)
		}
	}

	// Sorted, so two exports of the same subject list their sections
	// identically whatever order the wiring ran in.
	fmt.Println(registry.CollectorKeys())
}

Output:
[billing identity webhooks]

func NewRegistry 

func NewRegistry() *Registry

NewRegistry builds an empty Registry.

func (*Registry) Collector 

func (r *Registry) Collector(key string) (Collector, bool)

Collector returns the collector registered under key.

func (*Registry) CollectorKeys 

func (r *Registry) CollectorKeys() []string

CollectorKeys returns the registered collector keys, sorted.

Sorted rather than in registration order, so an artifact's manifest lists its sections identically whatever order the wiring happened to run in. Two exports of the same subject differing only in the order of a JSON array is the kind of diff that costs somebody an afternoon.

func (*Registry) Eraser 

func (r *Registry) Eraser(key string) (Eraser, bool)

Eraser returns the eraser registered under key.

func (*Registry) EraserKeys 

func (r *Registry) EraserKeys() []string

EraserKeys returns the registered eraser keys, sorted.

func (*Registry) RegisterCollector 

func (r *Registry) RegisterCollector(key string, collector Collector) error

RegisterCollector adds one domain's export collector under key, which becomes that domain's section name in every artifact.

Re-registering a key is an error rather than a replacement. A silent overwrite would drop a domain from every export from then on, and the only symptom would be a section missing from a file nobody reads until a regulator does.

func (*Registry) RegisterEraser 

func (r *Registry) RegisterEraser(key string, eraser Eraser) error

RegisterEraser adds one domain's eraser under key.

A domain may register a Collector, an Eraser, or both, and the keys are deliberately independent namespaces. A domain that holds data it must export but may never delete — an immutable ledger — registers only a collector, and that asymmetry is the normal case rather than a misconfiguration.

type Request 

type Request struct {
	// RequestedAt is when the request was submitted. It is the instant the
	// statutory clock starts, so it is stamped once and never rewritten — not
	// by a confirmation, and not by a retry.
	RequestedAt time.Time `json:"requestedAt"`

	// DueAt is when the response is legally owed, computed at submission from
	// the configured response window for the request type. See Overdue.
	DueAt time.Time `json:"dueAt"`

	// ExpiresAt is when the artifact is deleted and an export moves to
	// StatusExpired. For an erasure it is when the confirmation window lapses,
	// and it is zero once the erasure is confirmed.
	ExpiresAt time.Time `json:"expiresAt"`

	// CompletedAt is when the request reached a terminal state. Nil until it
	// does.
	CompletedAt *time.Time `json:"completedAt,omitempty"`

	// KeyShreddedAt is when this subject's data key was destroyed, for an
	// erasure fulfilled by a Fulfiller with a shredder configured. Nil otherwise,
	// which covers three different situations worth telling apart: no shredder
	// is wired, the request is scoped and therefore cannot shred, or the erasure
	// has not run yet. Retained says which of the first two it was.
	//
	// It is set even when there was no key to destroy. The claim it records is
	// not "bytes were overwritten" but "as of this instant no key exists for
	// this subject and none can be minted", which is the property the erasure
	// rests on either way.
	KeyShreddedAt *time.Time `json:"keyShreddedAt,omitempty"`

	// Failures records the collector or eraser keys that errored, against the
	// rendered error. A completed export with a non-empty Failures is a partial
	// export: the artifact was delivered, and its manifest names these same
	// sections as missing.
	//
	// It is a rendered string rather than an error because it is stored and read
	// by a human — often a regulator — not re-wrapped by a caller.
	Failures map[string]string `json:"failures,omitempty"`

	// Retained records, per eraser key, what an erasure kept and why. It is the
	// answer to "you said you deleted everything", and the reason ErasureOutcome
	// carries a legal basis rather than only a count.
	Retained map[string]string `json:"retained,omitempty"`

	// ID identifies the request.
	ID string `json:"id"`

	// OperationID names the operation fulfilling this request, and is what a
	// client polls for progress. Empty only while an erasure awaits
	// confirmation, because until somebody confirms it there is nothing running.
	//
	// It is a separate identifier rather than the request's own, and the two are
	// deliberately not made equal. They are different objects with different
	// lifetimes: the operation is reaped on operations.Config.Retention — weeks —
	// and this record is kept for years, so an ID that meant both would go on
	// resolving to one of them long after it stopped resolving to the other.
	OperationID string `json:"operationID,omitempty"`

	// ArtifactRef is the uploads path of the export artifact. Empty for an
	// erasure, for an incomplete export, and for an expired one — the path is
	// cleared when the object is deleted, so a stale reference cannot outlive
	// the thing it referenced.
	ArtifactRef string `json:"artifactRef,omitempty"`

	// LastError is why a failed request failed, rendered. Empty otherwise.
	//
	// The operation carries the same failure in a shape a client can branch on,
	// and carries it better — a stable code rather than a string. This is the
	// copy that survives the operation being reaped, for the record that has to
	// last three years.
	LastError string `json:"lastError,omitempty"`

	// Subject is who the request is about.
	Subject Subject `json:"subject"`

	// Type is what was asked for.
	Type RequestType `json:"type"`

	// Status is where it got to.
	Status Status `json:"status"`

	// ArtifactBytes is the stored size of the artifact, after compression and
	// encryption. Zero for an erasure or an unfulfilled export.
	ArtifactBytes int64 `json:"artifactBytes,omitempty"`

	// Deleted and Anonymized are the erasure totals summed across every eraser.
	Deleted    int64 `json:"deleted,omitempty"`
	Anonymized int64 `json:"anonymized,omitempty"`
}

Request is one export or erasure and everything known about how it went.

func (*Request) Overdue 

func (r *Request) Overdue(now time.Time) bool

Overdue reports whether the statutory response window has lapsed with the request still unfulfilled. A request that completed after its deadline is not overdue — it is late, which is a fact about the past and not a thing to page somebody about.

func (*Request) Partial 

func (r *Request) Partial() bool

Partial reports whether a completed request left something uncollected or unerased. A caller rendering a status page should say so: "completed" over a manifest with three missing sections is a misleading thing to show a subject who has thirty days to complain about it.

type RequestType 

type RequestType string

RequestType names what a request asks for. 

const (
	// RequestExport is a subject access request: collect everything held about
	// the subject and deliver it.
	RequestExport RequestType = "export"
	// RequestErasure is a right-to-be-forgotten request: delete or anonymize
	// what is held about the subject, retaining only what must be retained.
	RequestErasure RequestType = "erasure"
)

func (RequestType) Valid 

func (t RequestType) Valid() bool

Valid reports whether t is a request type this package implements.

type SQLStore 

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

SQLStore is the SQL-backed Store, against the schema dataprivacy/migrations renders.

It is exported, and returned by NewSQLStore, so a caller who has chosen SQL storage can depend on that choice rather than on the Store seam every backing shares.

func NewSQLStore 

func NewSQLStore(client database.Client, opts ...SQLStoreOption) (*SQLStore, error)

NewSQLStore builds a Store over the given database.

The dialect comes from the client, so the two cannot disagree. The prefix must still match the one the migrations were rendered with — nothing here can check that, and a mismatch surfaces as a missing table on the first query rather than at construction.

Observability is optional and defaults to nothing: an unconfigured store logs to a noop logger, traces to a noop provider, and counts into a noop meter.

func (*SQLStore) CompleteErasure 

func (s *SQLStore) CompleteErasure(ctx context.Context, q database.SQLQueryExecutor, req *Request, at time.Time) error

func (*SQLStore) CompleteExport 

func (s *SQLStore) CompleteExport(ctx context.Context, q database.SQLQueryExecutor, req *Request, at time.Time) error

func (*SQLStore) CountOverdue 

func (s *SQLStore) CountOverdue(ctx context.Context, now time.Time) (map[RequestType]int64, error)

func (*SQLStore) ExpiringArtifacts 

func (s *SQLStore) ExpiringArtifacts(ctx context.Context, now time.Time, limit int) ([]*Request, error)

func (*SQLStore) Fail 

func (s *SQLStore) Fail(ctx context.Context, requestID, lastErr string, at time.Time) (bool, error)

func (*SQLStore) Get 

func (s *SQLStore) Get(ctx context.Context, requestID string) (*Request, error)

func (*SQLStore) LapseUnconfirmed 

func (s *SQLStore) LapseUnconfirmed(ctx context.Context, now time.Time, limit int) (int64, error)

func (*SQLStore) List 

func (s *SQLStore) List(
	ctx context.Context,
	subject Subject,
	filter *filtering.QueryFilter,
) (*filtering.QueryFilteredResult[Request], error)

func (*SQLStore) MarkExpired 

func (s *SQLStore) MarkExpired(ctx context.Context, requestID string, at time.Time) error

func (*SQLStore) MarkKeyShredded 

func (s *SQLStore) MarkKeyShredded(ctx context.Context, requestID string, at time.Time) error

func (*SQLStore) Reap 

func (s *SQLStore) Reap(ctx context.Context, before time.Time, limit int) (int64, error)

func (*SQLStore) Save 

func (s *SQLStore) Save(ctx context.Context, q database.SQLQueryExecutor, req *Request) error

func (*SQLStore) Transition 

func (s *SQLStore) Transition(
	ctx context.Context,
	q database.SQLQueryExecutor,
	requestID string,
	from []Status,
	to Status,
	operationID string,
	at time.Time,
) (*Request, error)

func (*SQLStore) WithTransaction 

func (s *SQLStore) WithTransaction(ctx context.Context, fn func(q database.SQLQueryExecutor) error) error

WithTransaction delegates to the client, which begins its own span for the transaction. Wrapping it here would nest a second span around the first and say nothing the client's does not.

type SQLStoreOption 

type SQLStoreOption func(*SQLStore)

SQLStoreOption configures a SQL Store.

func WithStoreLogger 

func WithStoreLogger(logger logging.Logger) SQLStoreOption

WithStoreLogger attaches a logger.

func WithStoreMetricsProvider 

func WithStoreMetricsProvider(metricsProvider metrics.Provider) SQLStoreOption

WithStoreMetricsProvider attaches a metrics provider.

func WithStoreTracerProvider 

func WithStoreTracerProvider(tracerProvider tracing.Provider) SQLStoreOption

WithStoreTracerProvider attaches a tracer provider.

func WithTablePrefix 

func WithTablePrefix(prefix string) SQLStoreOption

WithTablePrefix overrides DefaultTablePrefix. It must be a plain SQL identifier fragment: it is interpolated into the query text, not bound as a parameter, and it must match the prefix the migrations were rendered with.

type Service 

type Service interface {
	// Submit records a new request, starts the operation that fulfills it, and
	// returns the request with Request.OperationID set.
	//
	// The row and the operation are written in one transaction, so a process
	// that dies between them leaves neither. The enqueue that follows is not,
	// and cannot be — see operations.Service.StartInTransaction — so a request
	// submitted at exactly the wrong moment waits for the operations recovery
	// sweep rather than for a worker. It is recorded and readable throughout.
	//
	// An erasure submitted to a Service with a confirmation window returns
	// StatusAwaitingConfirmation and an empty OperationID, and nothing runs
	// until Confirm.
	Submit(ctx context.Context, subject Subject, t RequestType) (*Request, error)

	// Get reads one request. It returns an error wrapping ErrRequestNotFound
	// when there is no such request.
	Get(ctx context.Context, requestID string) (*Request, error)

	// List pages through a subject's requests. A subject is entitled to know
	// what has been asked in their name, which is the reason this is scoped to
	// a subject rather than global.
	//
	// Ordering follows the filter's SortBy — ascending by default, as
	// filtering.DefaultQueryFilter asks. Requests are ordered by ID, which for
	// generated identifiers is submission order.
	List(ctx context.Context, subject Subject, f *filtering.QueryFilter) (*filtering.QueryFilteredResult[Request], error)

	// Confirm moves an erasure out of StatusAwaitingConfirmation and starts the
	// operation that fulfills it, returning the request with OperationID set.
	// It returns an error wrapping ErrNotAwaitingConfirmation for a request in
	// any other state, including one whose window has already lapsed.
	Confirm(ctx context.Context, requestID string) (*Request, error)

	// Cancel withdraws a request.
	//
	// An unconfirmed erasure is cancelled outright: nothing has begun and there
	// is nothing to unwind. A request already in progress has its operation
	// asked to stop, which is a request rather than a kill — the runner stops
	// between domains, at a point it can describe, and marks this row cancelled
	// when it does. So Cancel on an in-progress request returns it still
	// StatusInProgress, and the operation is where the answer arrives.
	//
	// An erasure that has begun erasing may finish anyway, and that is the
	// honest outcome rather than a gap: the erasers share one transaction and
	// the shred that precedes them cannot be undone, so the last moment at which
	// stopping means anything is before either has run.
	Cancel(ctx context.Context, requestID string) (*Request, error)

	// Download mints a time-limited URL for a completed export's artifact,
	// letting the subject fetch it from storage without the bytes passing
	// through the application. The URL expires; the artifact behind it is
	// deleted at ExpiresAt whether or not anyone fetched it.
	//
	// It returns an error wrapping ErrArtifactUnavailable for a request with no
	// artifact, ErrArtifactEncrypted when the Service encrypts artifacts at
	// rest, and ErrNoURLSigner when the storage provider cannot sign.
	Download(ctx context.Context, requestID string) (string, error)

	// Open returns a completed export's artifact as canonical JSON, reversing
	// whatever compression and encryption it was stored under. The caller must
	// close it.
	//
	// It does not stream, despite returning a reader: decryption and
	// decompression both need the whole object, so the artifact is read into
	// memory in full and the reader hands it back. Sizing follows from that —
	// see DefaultMaxDocumentBytes for the ceiling an export is built under.
	//
	// It is the path that always works — every storage provider, encrypted or
	// not — at the cost of proxying the bytes through the application. Prefer
	// Download where it is available, and reach for this when it is not.
	Open(ctx context.Context, requestID string) (io.ReadCloser, error)
}

Service is the application-facing seam: submit a request, ask after one, list them.

Fulfillment is deliberately not on this interface. A Submit that collected eleven domains inline would tie a regulatory obligation to the lifetime of an HTTP request, and the one guarantee a subject access request needs is that it survives the process that accepted it. Submit writes a row and starts an operation; an operations Worker runs it.

There is no status endpoint here either, and there deliberately is not one. "How far along is my export" is answered by operations/http against Request.OperationID — the same endpoint, the same shape, and the same event stream every other long-running thing in the application already uses.

type ServiceConfig 

type ServiceConfig struct {
	// ExportResponseWindow is how long an export may take before it counts as
	// overdue. Defaults to DefaultResponseWindow.
	ExportResponseWindow time.Duration `env:"EXPORT_RESPONSE_WINDOW" json:"exportResponseWindow,omitempty" yaml:"exportResponseWindow,omitempty"`

	// ErasureResponseWindow is the same for erasures. Separate from the export
	// window because the jurisdictions that distinguish them give erasure the
	// longer one, and a single knob would force the stricter deadline onto both.
	ErasureResponseWindow time.Duration `env:"ERASURE_RESPONSE_WINDOW" json:"erasureResponseWindow,omitempty" yaml:"erasureResponseWindow,omitempty"`

	// ConfirmationWindow is how long an erasure waits for confirmation before
	// it is cancelled. Zero — the default — means erasures are queued on
	// submission and Confirm is never needed.
	//
	// Turning it on is the difference between an accidental erasure being a
	// support ticket and being unrecoverable. Regulation generally permits a
	// verification step, and the failure mode it prevents is the only one in
	// this package that cannot be undone.
	ConfirmationWindow time.Duration `env:"CONFIRMATION_WINDOW" json:"confirmationWindow,omitempty" yaml:"confirmationWindow,omitempty"`

	// SignedURLTTL is how long a download URL is valid. Defaults to
	// DefaultSignedURLTTL.
	SignedURLTTL time.Duration `env:"SIGNED_URL_TTL" json:"signedURLTTL,omitempty" yaml:"signedURLTTL,omitempty"`
}

ServiceConfig configures the request state machine's timings.

func (*ServiceConfig) EnsureDefaults 

func (cfg *ServiceConfig) EnsureDefaults()

EnsureDefaults fills unset knobs with the package defaults.

func (*ServiceConfig) ValidateWithContext 

func (cfg *ServiceConfig) ValidateWithContext(ctx context.Context) error

ValidateWithContext validates a ServiceConfig.

type ServiceOption 

type ServiceOption func(*StoreService)

ServiceOption configures a Service.

func WithActorResolver 

func WithActorResolver(resolver ActorResolver) ServiceOption

WithActorResolver supplies the principal recorded in audit entries.

func WithServiceAuditRecorder 

func WithServiceAuditRecorder(recorder audit.Recorder) ServiceOption

WithServiceAuditRecorder attaches the audit log this package writes to.

Every submission and every state change it drives is recorded. That is not decoration: an export artifact is the most sensitive object an application produces, and a system that can produce one without leaving a record of who asked has a data exfiltration path with no alarm on it.

func WithServiceClock 

func WithServiceClock(c clock.Clock) ServiceOption

WithServiceClock swaps the clock stamping submission, deadline, and expiry.

func WithServiceCompressor 

func WithServiceCompressor(compressor compression.Compressor) ServiceOption

WithServiceCompressor supplies the compressor artifacts were written with. It must match the Fulfiller's, or Open returns garbage.

func WithServiceDecryptor 

func WithServiceDecryptor(decryptor encryption.Decryptor) ServiceOption

WithServiceDecryptor supplies the decryptor for artifacts written encrypted. It must match the Fulfiller's encryptor.

Setting it also disables Download: see ErrArtifactEncrypted.

func WithServiceLogger 

func WithServiceLogger(logger logging.Logger) ServiceOption

WithServiceLogger attaches a logger.

func WithServiceMetricsProvider 

func WithServiceMetricsProvider(metricsProvider metrics.Provider) ServiceOption

WithServiceMetricsProvider attaches a metrics provider.

func WithServiceTracerProvider 

func WithServiceTracerProvider(tracerProvider tracing.Provider) ServiceOption

WithServiceTracerProvider attaches a tracer provider.

func WithServiceUploadManager 

func WithServiceUploadManager(manager uploads.UploadManager) ServiceOption

WithServiceUploadManager supplies the storage artifacts are read from.

It must be the same storage, and the same path prefix, that the Fulfiller writes to. Nothing here can check that, and a mismatch surfaces as an artifact that exists in the bucket and cannot be found by the service that promised it.

type Status 

type Status string

Status is where a request has got to.

The transitions between these are diagrammed in the package overview.

It is not the operation's state and does not mirror it. The operation says how the current attempt is going — pending, running, how many units in, which attempt — and is reaped on its own retention. This says what the request is, as the statutory record of it: whether somebody still has to confirm it, whether it was fulfilled, and whether the artifact it produced still exists. Those are different questions with different lifetimes, and collapsing them into one column would have the record of a request somebody made three years ago disappear along with the progress bar.

The one state to dwell on is expired. It is reachable only from completed, and only for an export, and it is the state people forget. An export artifact contains everything an application knows about a person; without an expiry it is a permanent object in a bucket. 

const (
	// StatusAwaitingConfirmation is an erasure that has been submitted but not
	// yet confirmed. Reachable only when a confirmation window is configured,
	// and the one state in which no operation exists yet.
	StatusAwaitingConfirmation Status = "awaiting_confirmation"
	// StatusInProgress is a request an operation is fulfilling.
	//
	// It covers what used to be two states, pending and processing, because the
	// difference between them is now the operation's to record and this row
	// genuinely does not know it: whether a worker has picked the operation up,
	// which attempt it is on, and how far through the domains it has got are all
	// answers Request.OperationID points at.
	StatusInProgress Status = "in_progress"
	// StatusCompleted is a request that was fulfilled. An export in this state
	// has an ArtifactRef; it may also have Failures, which is what a partial
	// export looks like.
	StatusCompleted Status = "completed"
	// StatusFailed is a request whose operation gave up: it exhausted its
	// attempts, or failed for a reason no retry would fix.
	//
	// It is written by the runner on its final attempt rather than by the
	// operations worker, which knows the operation failed and has no notion of
	// the request behind it. See operations.Attempt.
	StatusFailed Status = "failed"
	// StatusExpired is a completed export whose artifact has been deleted.
	StatusExpired Status = "expired"
	// StatusCancelled is a request that was withdrawn: an erasure nobody
	// confirmed before its window lapsed, one the subject cancelled, or one
	// stopped mid-flight through the operation's cancellation.
	StatusCancelled Status = "cancelled"
)

func (Status) Terminal 

func (s Status) Terminal() bool

Terminal reports whether a status is one nothing moves out of.

func (Status) Valid 

func (s Status) Valid() bool

Valid reports whether s is a status this package writes.

type Store 

type Store interface {
	// Save inserts a new request using the caller's executor. It does not
	// update: a request row's history is the thing being recorded, and an upsert
	// here would let a resubmission quietly overwrite the timestamp the
	// statutory clock runs from.
	//
	// It takes an executor for the same reason audit.Recorder.Record does. "Who
	// asked for this person's data" is itself an auditable event, and an audit
	// entry that can commit while the request it describes rolls back — or the
	// reverse — is not a record of anything.
	Save(ctx context.Context, q database.SQLQueryExecutor, req *Request) error

	// Get reads one request. It returns an error wrapping ErrRequestNotFound
	// when there is no such request.
	Get(ctx context.Context, requestID string) (*Request, error)

	// List pages through a subject's requests, ordered by ID in the direction
	// the filter's SortBy asks for.
	List(ctx context.Context, subject Subject, filter *filtering.QueryFilter) (*filtering.QueryFilteredResult[Request], error)

	// Transition moves a request from any of the `from` statuses to `to` using
	// the caller's executor, returning the updated request. It returns an error
	// wrapping ErrRequestNotFound when no row matched — which covers both "no
	// such request" and "the request was not in a state this transition applies
	// to", so callers wrap it into whichever of the two their API means.
	//
	// operationID is recorded alongside the new status when it is non-empty,
	// because the one transition that sets it — a confirmation, which starts the
	// operation as it moves the row — must not be able to commit the status
	// without the pointer to the thing now doing the work.
	Transition(
		ctx context.Context,
		q database.SQLQueryExecutor,
		requestID string,
		from []Status,
		to Status,
		operationID string,
		at time.Time,
	) (*Request, error)

	// CompleteExport records a fulfilled export using the caller's executor: its
	// artifact, that artifact's expiry, and any per-section failures.
	CompleteExport(ctx context.Context, q database.SQLQueryExecutor, req *Request, at time.Time) error

	// WithTransaction runs fn against the store's database.
	//
	// It is on this interface because an erasure has to be atomic across
	// domains and with its own bookkeeping: every registered Eraser and the
	// request's completion share one transaction, so a subject is never left
	// half-erased across eleven domains because the ninth failed. A Store that
	// is not backed by the same database as the erasers cannot offer that, and
	// should refuse erasure rather than pretend.
	WithTransaction(ctx context.Context, fn func(q database.SQLQueryExecutor) error) error

	// CompleteErasure records a fulfilled erasure using the caller's executor,
	// so it commits with the deletions it describes.
	CompleteErasure(ctx context.Context, q database.SQLQueryExecutor, req *Request, at time.Time) error

	// MarkKeyShredded records that the subject's data key was destroyed, on its
	// own and before the erasure it belongs to has finished.
	//
	// It is separate from CompleteErasure because the destruction is separate.
	// It is irreversible, it happens before any row is deleted, and a request
	// that then exhausts its attempts has still destroyed the key — so writing
	// it only at completion would leave the one fact about an erasure that
	// nothing else can reconstruct recorded nowhere.
	//
	// It is idempotent. A retried erasure re-shreds, gets the original
	// destruction time back, and must not overwrite the record with a later one.
	MarkKeyShredded(ctx context.Context, requestID string, at time.Time) error

	// Fail moves an in-progress request to StatusFailed, recording why, and
	// reports whether it moved anything.
	//
	// It is called only on an operation's final attempt — see
	// operations.Attempt — because that is the only moment at which "this
	// request will not be fulfilled" is a true thing to write. Every earlier
	// failure leaves the row in StatusInProgress, which is what it is: the
	// operation is going to try again.
	//
	// False with a nil error means the row was not in StatusInProgress: it was
	// cancelled, or completed by a duplicate execution that got there first. It
	// is not an error, because in both of those the row already says something
	// truer than "failed" — but the caller has to know, because telling a
	// subject their request failed when it was cancelled is worse than telling
	// them nothing.
	Fail(ctx context.Context, requestID, lastErr string, at time.Time) (bool, error)

	// ExpiringArtifacts returns completed exports whose artifacts are due for
	// deletion. The sweeper deletes each object before calling MarkExpired, so
	// this deliberately returns the requests rather than expiring them in bulk:
	// a row marked expired while its object survived is a file nobody is
	// looking for any more and nobody will delete.
	ExpiringArtifacts(ctx context.Context, now time.Time, limit int) ([]*Request, error)

	// MarkExpired clears a request's artifact reference and moves it to
	// StatusExpired, once the object itself is gone.
	MarkExpired(ctx context.Context, requestID string, at time.Time) error

	// LapseUnconfirmed cancels erasures whose confirmation window has passed,
	// returning how many were cancelled.
	LapseUnconfirmed(ctx context.Context, now time.Time, limit int) (int64, error)

	// CountOverdue counts unfulfilled requests past their statutory deadline,
	// by request type, for the sweeper's gauge.
	CountOverdue(ctx context.Context, now time.Time) (map[RequestType]int64, error)

	// Reap deletes terminal request records completed before the given time, up
	// to limit rows.
	//
	// Records of privacy requests are themselves personal data, and keeping
	// them forever is the mistake this package would otherwise make on every
	// consumer's behalf. What it does not do is delete a request whose artifact
	// still exists — see the retention discussion in the package docs.
	Reap(ctx context.Context, before time.Time, limit int) (int64, error)
}

Store is the persistence seam for the request state machine.

This package ships a SQL implementation (NewSQLStore) together with the DDL it needs (dataprivacy/migrations), so adopting it does not mean writing this. The interface exists because the state machine and its storage are genuinely separable, and an application with its own schema conventions should not have to fork the package to keep them.

Every transition method is a conditional write rather than a read-then-write. Two workers can claim, a sweeper can expire, and a subject can cancel, all at the same instant; a store that read the row, decided, and wrote it back would resolve those races by whichever transaction was slower. The predicates are in the queries for that reason, and a transition that matched nothing returns an error rather than silently succeeding.

type StoreService 

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

StoreService is the request state machine, over a Store and an operations Service. It is exported, and returned by NewService, so a caller can depend on the service it built rather than on the Service seam.

func NewService 

func NewService(
	ctx context.Context,
	cfg *ServiceConfig,
	store Store,
	ops operations.Service,
	opts ...ServiceOption,
) (*StoreService, error)

NewService builds a Service.

ops is where the work goes. It is a required argument rather than an option because a Service without one would record requests that nothing ever fulfills — which looks exactly like a working Service until a subject's statutory window runs out.

It must be an operations Service whose registry has this package's kinds registered, which is what Fulfiller.Register does. That is true even on a process that only submits: operations.Service.Start resolves the kind at submission so that an unrunnable operation is refused there rather than discovered in a worker an hour later.

ctx is used to validate the config and is not retained.

func (*StoreService) Cancel 

func (s *StoreService) Cancel(ctx context.Context, requestID string) (*Request, error)

func (*StoreService) Confirm 

func (s *StoreService) Confirm(ctx context.Context, requestID string) (*Request, error)

func (*StoreService) Download 

func (s *StoreService) Download(ctx context.Context, requestID string) (string, error)

func (*StoreService) Get 

func (s *StoreService) Get(ctx context.Context, requestID string) (*Request, error)

func (*StoreService) List 

func (s *StoreService) List(
	ctx context.Context,
	subject Subject,
	filter *filtering.QueryFilter,
) (*filtering.QueryFilteredResult[Request], error)

func (*StoreService) Open 

func (s *StoreService) Open(ctx context.Context, requestID string) (io.ReadCloser, error)

func (*StoreService) Submit 

func (s *StoreService) Submit(ctx context.Context, subject Subject, t RequestType) (*Request, error)

type Subject 

type Subject struct {
	// ID identifies the subject. Required.
	ID string `json:"id"`
	// Scope is the account or tenant the request is confined to, when it is
	// confined at all. Empty means the request spans every scope the subject
	// appears in, which is what a plain "give me my data" asks for.
	//
	// It is one opaque string rather than a typed tenancy path for the same
	// reason audit.Entry.Scope is: tenancy depth is an application's decision,
	// and a two-level model cannot express one level or three.
	Scope string `json:"scope,omitempty"`
	// Type says what kind of subject it is.
	Type SubjectType `json:"type,omitempty"`
}

Subject is who or what a request is about.

type SubjectType 

type SubjectType string

SubjectType distinguishes the kinds of thing a request can be about.

Like audit.ActorType this is a bare string with suggested constants rather than a closed set: an application whose data hangs off a third kind of principal should say so rather than misfile it as one of these. 

const (
	// SubjectUser is a natural person — the subject GDPR and CCPA are written
	// about.
	SubjectUser SubjectType = "user"
	// SubjectAccount is an account, tenant, or organization. An account-scoped
	// request is the one that arrives when a business customer leaves.
	SubjectAccount SubjectType = "account"
)

type SweepResult 

type SweepResult struct {
	// Overdue is how many unfulfilled requests are past their deadline, by
	// type. Reported whether or not anything was swept — it is the number an
	// operator actually needs, and a sweep that found nothing to delete is
	// exactly when nobody would otherwise look.
	Overdue map[RequestType]int64
	// ArtifactsExpired is how many artifacts were deleted from storage.
	ArtifactsExpired int64
	// ErasuresLapsed is how many unconfirmed erasures were cancelled.
	ErasuresLapsed int64
	// RecordsReaped is how many terminal request records were deleted.
	RecordsReaped int64
}

SweepResult is what one pass did.

type Sweeper 

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

Sweeper runs the three background chores this package needs: deleting expired artifacts, cancelling erasures nobody confirmed, and reaping request records past retention. It also samples the overdue gauge.

The artifact expiry is the one that matters. Everything else here is housekeeping; that one is the difference between an export being a temporary artifact and being a permanent object in a bucket containing everything an application knows about a person. A deployment that fulfills requests and does not run the Sweeper accumulates those forever, which is why this is a separate, named, schedulable thing rather than a flag on the Fulfiller.

func NewSweeper 

func NewSweeper(ctx context.Context, cfg *SweeperConfig, store Store, opts ...SweeperOption) (*Sweeper, error)

NewSweeper builds a Sweeper. It does not schedule it; see Job.

ctx is used to validate the config and is not retained.

func (*Sweeper) Job 

func (s *Sweeper) Job(schedule jobs.Schedule, leaseTTL time.Duration) jobs.Job

Job renders the Sweeper as a jobs.Job, for registration with a jobs.Scheduler.

Scheduling it there rather than running a ticker of its own is what makes the sweep run once across a fleet instead of once per replica. Ten replicas each deleting the same artifacts is mostly harmless; ten replicas each reaping the same rows is ten times the lock contention on a table people are still reading.

LeaseTTL must comfortably exceed one sweep. The scheduler does not renew a lease while a job runs, so a sweep that outlives its lease loses exclusivity halfway through — see jobs.Job.LeaseTTL.

func (*Sweeper) Sweep 

func (s *Sweeper) Sweep(ctx context.Context) (*SweepResult, error)

Sweep runs one pass: lapse, expire, reap, and sample.

The four run in that order and independently. An error in one is recorded and the rest still run — they are unrelated chores sharing a schedule, and a storage provider being unreachable is not a reason to skip the retention reap as well.

type SweeperConfig 

type SweeperConfig struct {
	// RequestRetention is how long a terminal request record is kept. Defaults
	// to DefaultRequestRetention.
	RequestRetention time.Duration `env:"REQUEST_RETENTION" json:"requestRetention,omitempty" yaml:"requestRetention,omitempty"`

	// BatchSize caps how much one sweep tick does, so a long-neglected table is
	// trimmed over several passes instead of one statement that holds locks for
	// minutes.
	BatchSize int `env:"BATCH_SIZE" json:"batchSize,omitempty" yaml:"batchSize,omitempty"`

	// DisableReap stops the sweeper deleting terminal request records.
	//
	// It exists because "how long do we keep the record that somebody asked" is
	// a jurisdiction's answer and not a library's, and an operator whose answer
	// is "forever, and we will argue about it later" should be able to say so
	// without setting a retention of a hundred years.
	DisableReap bool `env:"DISABLE_REAP" json:"disableReap,omitempty" yaml:"disableReap,omitempty"`
}

SweeperConfig configures the expiry, lapse, and retention sweeps.

func (*SweeperConfig) EnsureDefaults 

func (cfg *SweeperConfig) EnsureDefaults()

EnsureDefaults fills unset knobs with the package defaults.

func (*SweeperConfig) ValidateWithContext 

func (cfg *SweeperConfig) ValidateWithContext(ctx context.Context) error

ValidateWithContext validates a SweeperConfig.

type SweeperOption 

type SweeperOption func(*Sweeper)

SweeperOption configures a Sweeper.

func WithSweeperClock 

func WithSweeperClock(c clock.Clock) SweeperOption

WithSweeperClock swaps the clock deciding what has expired.

func WithSweeperLogger 

func WithSweeperLogger(logger logging.Logger) SweeperOption

WithSweeperLogger attaches a logger.

func WithSweeperMetricsProvider 

func WithSweeperMetricsProvider(metricsProvider metrics.Provider) SweeperOption

WithSweeperMetricsProvider attaches a metrics provider, enabling the overdue gauge — which is the one instrument in this package worth alerting on.

func WithSweeperTracerProvider 

func WithSweeperTracerProvider(tracerProvider tracing.Provider) SweeperOption

WithSweeperTracerProvider attaches a tracer provider.

func WithSweeperUploadManager 

func WithSweeperUploadManager(manager uploads.UploadManager) SweeperOption

WithSweeperUploadManager supplies the storage artifacts are deleted from. It must be the same storage the Fulfiller writes to.

Without it the Sweeper refuses to expire artifacts at all rather than marking rows expired against objects it cannot delete — a row that says the artifact is gone while the artifact is not is worse than no sweep, because it stops anybody looking.

type URLSignerOption 

type URLSignerOption func(*clock.Clock)

URLSignerOption configures the signer NewArtifactURLSigner returns.

It is its own type rather than a FulfillerOption because the signer is built before the Fulfiller it is handed to, and is equally usable by a caller that has no Fulfiller at all.

func WithURLSignerClock 

func WithURLSignerClock(c clock.Clock) URLSignerOption

WithURLSignerClock swaps the clock the signer stamps its expiry against, so a Fulfiller under a test clock and the notification it sends agree about when the link stops working. An absent clock reads the wall clock.

Source Files

View all Source files

Directories

Path Synopsis
Package auditerasure supplies a dataprivacy.Eraser for the audit log.
 Package auditerasure supplies a dataprivacy.Eraser for the audit log.
Package dataprivacycfg assembles the data privacy machinery from environment configuration: the Store every part shares, the Service applications submit through, the Fulfiller that does the work, and the Sweeper that expires.
 Package dataprivacycfg assembles the data privacy machinery from environment configuration: the Store every part shares, the Service applications submit through, the Fulfiller that does the work, and the Sweeper that expires.
Package migrations supplies the data-privacy request table's DDL, rendered for a dialect and table prefix.
 Package migrations supplies the data-privacy request table's DDL, rendered for a dialect and table prefix.
Package dataprivacymock provides moq-generated mock implementations of interfaces in the dataprivacy package.
 Package dataprivacymock provides moq-generated mock implementations of interfaces in the dataprivacy package.

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.