feat(store): add the principal store and share-locked reads

internal/testutil/postgres.go

@@ -1,6 +1,7 @@
 package testutil
 
 import (
+	"runtime"
 	"testing"
 	"time"
 
@@ -52,3 +53,17 @@ func CreatePostgres(t *testing.T) *pgxpool.Pool {
 	require.NoError(t, migrations.Up(ctx, pool, zap.NewNop()))
 	return pool
 }
+
+// PostgresOrSkip is [CreatePostgres] for a test that runs only where Docker
+// is: without it the test skips, except on a Linux CI runner, where Docker is
+// expected and its absence fails the test.
+func PostgresOrSkip(t *testing.T) *pgxpool.Pool {
+	t.Helper()
+	if IsRunningInCI(t) && runtime.GOOS == "linux" && !IsDockerAvailable(t) {
+		t.Fatalf("docker is expected in CI linux testing environments, but wasn't found")
+	}
+	if !IsDockerAvailable(t) {
+		t.SkipNow()
+	}
+	return CreatePostgres(t)
+}

internal/testutil/wait.go

@@ -0,0 +1,48 @@
+package testutil
+
+import (
+	"testing"
+	"time"
+)
+
+// MinBlock is how long a call that must wait on a writer has to stay blocked.
+const MinBlock = 300 * time.Millisecond
+
+// WaitTimeout is how long a call that must finish is given before the test fails.
+const WaitTimeout = 10 * time.Second
+
+// RequireWaitsForWriter runs write in the background; write must close
+// entered once it holds its lock, then block until release is closed before
+// finishing. It then starts wait, requires that wait has not returned within [MinBlock],
+// releases the writer, and returns the writer's and wait's errors.
+func RequireWaitsForWriter(t *testing.T, write func(entered chan<- struct{}, release <-chan struct{}) error, wait func() error) (writeErr, waitErr error) {
+	t.Helper()
+	entered := make(chan struct{})
+	release := make(chan struct{})
+	wrote := make(chan error, 1)
+	go func() { wrote <- write(entered, release) }()
+	select {
+	case <-entered:
+	case err := <-wrote:
+		t.Fatalf("the writer returned before taking its lock: %v", err)
+	case <-time.After(WaitTimeout):
+		t.Fatal("the writer did not take its lock")
+	}
+
+	waited := make(chan error, 1)
+	go func() { waited <- wait() }()
+	select {
+	case <-waited:
+		t.Fatal("the waiting call returned while the writer held its lock")
+	case <-time.After(MinBlock):
+	}
+
+	close(release)
+	writeErr = <-wrote
+	select {
+	case waitErr = <-waited:
+	case <-time.After(WaitTimeout):
+		t.Fatal("the waiting call did not return after the writer finished")
+	}
+	return writeErr, waitErr
+}

pkg/api/service/tenant/service.go

@@ -15,6 +15,7 @@ import (
 	"github.com/fil-forge/hilt/pkg/store/accesskey"
 	"github.com/fil-forge/hilt/pkg/store/bucket"
 	"github.com/fil-forge/hilt/pkg/store/delegation"
+	principalstore "github.com/fil-forge/hilt/pkg/store/principal"
 	"github.com/fil-forge/hilt/pkg/store/provider"
 	tenantstore "github.com/fil-forge/hilt/pkg/store/tenant"
 	wrapkeystore "github.com/fil-forge/hilt/pkg/store/wrapkey"
@@ -38,6 +39,7 @@ type Service struct {
 	providers   provider.Store
 	buckets     bucket.Store
 	accessKeys  accesskey.Store
+	principals  principalstore.Store
 	delegations delegation.Store
 	secrets     vault.Vault
 	wrapKeys    wrapkeystore.Store
@@ -52,6 +54,7 @@ func New(
 	providers provider.Store,
 	buckets bucket.Store,
 	accessKeys accesskey.Store,
+	principals principalstore.Store,
 	delegations delegation.Store,
 	secrets vault.Vault,
 	wrapKeys wrapkeystore.Store,
@@ -64,6 +67,7 @@ func New(
 		providers:   providers,
 		buckets:     buckets,
 		accessKeys:  accessKeys,
+		principals:  principals,
 		delegations: delegations,
 		secrets:     secrets,
 		plcClient:   plcClient,
@@ -255,8 +259,8 @@ func (s *Service) SetStatus(ctx context.Context, externalID, status string) erro
 }
 
 // Delete permanently deletes a tenant (which must be disabled), cascading to its
-// buckets, access keys, and delegations, and deactivating its did:plc. It is
-// idempotent: a missing tenant is a no-op.
+// buckets, access keys, principals, and delegations, and deactivating its
+// did:plc. It is idempotent: a missing tenant is a no-op.
 //
 // Out of scope: deprovisioning the tenant's spaces from the Forge upload service
 // (Sprue), for which there is no facility per the RFC.
@@ -299,6 +303,13 @@ func (s *Service) Delete(ctx context.Context, externalID string) error {
 		}
 	}
 
+	// Cascade: principals. They reference the tenant row (FK RESTRICT), so they
+	// go before it. Nothing is published: the tenant is disabled and its keys are
+	// gone, so no request can reach a principal's cached access.
+	if err := s.principals.DeleteByTenant(ctx, rec.ID); err != nil {
+		return fmt.Errorf("deleting principals: %w", err)
+	}
+
 	// Cascade: wrap keys (registry rows + their sealed private halves). Every
 	// version, active and archived, is destroyed — tenant deletion is the
 	// "true deletion" that ends the envelope-recovery path, so archive-don't-

pkg/api/service/tenant/service_test.go

@@ -15,6 +15,7 @@ import (
 	accesskeymemory "github.com/fil-forge/hilt/pkg/store/accesskey/memory"
 	bucketmemory "github.com/fil-forge/hilt/pkg/store/bucket/memory"
 	delegationmemory "github.com/fil-forge/hilt/pkg/store/delegation/memory"
+	principalmemory "github.com/fil-forge/hilt/pkg/store/principal/memory"
 	providermemory "github.com/fil-forge/hilt/pkg/store/provider/memory"
 	"github.com/fil-forge/hilt/pkg/store/tenant"
 	tenantmemory "github.com/fil-forge/hilt/pkg/store/tenant/memory"
@@ -83,7 +84,7 @@ func provisionSetup(t *testing.T, plcStatus int) provisionEnv {
 	require.NoError(t, err)
 
 	svc := tenantsvc.New(zap.NewNop(), tenantmemory.New(), providers, bucketmemory.New(),
-		accesskeymemory.New(), delegationmemory.New(), vaultmemory.New(), wrapkeysmemory.New(), plcClient, upload)
+		accesskeymemory.New(), principalmemory.New(), delegationmemory.New(), vaultmemory.New(), wrapkeysmemory.New(), plcClient, upload)
 	return provisionEnv{svc: svc, providers: providers, sprueFailed: sprueFailed}
 }
 
@@ -144,7 +145,7 @@ func TestProvision(t *testing.T) {
 // clients — enough for Get and SetStatus, which never touch them.
 func simpleService(tenants tenant.Store) *tenantsvc.Service {
 	return tenantsvc.New(zap.NewNop(), tenants, providermemory.New(), bucketmemory.New(),
-		accesskeymemory.New(), delegationmemory.New(), vaultmemory.New(), wrapkeysmemory.New(), nil, nil)
+		accesskeymemory.New(), principalmemory.New(), delegationmemory.New(), vaultmemory.New(), wrapkeysmemory.New(), nil, nil)
 }
 
 func TestGetAndSetStatus(t *testing.T) {
@@ -215,9 +216,11 @@ func (d *plcDirectory) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 }
 
 type deleteEnv struct {
-	svc       *tenantsvc.Service
-	tenants   *tenantmemory.Store
-	directory *plcDirectory
+	svc        *tenantsvc.Service
+	tenants    *tenantmemory.Store
+	principals *principalmemory.Store
+	tenantID   did.DID
+	directory  *plcDirectory
 }
 
 func deleteSetup(t *testing.T, status tenant.Status) deleteEnv {
@@ -245,12 +248,14 @@ func deleteSetup(t *testing.T, status tenant.Status) deleteEnv {
 
 	tenants := tenantmemory.New()
 	secrets := vaultmemory.New()
+	principals := principalmemory.New()
 	require.NoError(t, tenants.Add(ctx, tenantID, "tenant-1", testutil.RandomDID(t), status))
 	require.NoError(t, secrets.Write(ctx, vault.TenantKeyPath(tenantID), signer.Bytes()))
+	require.NoError(t, principals.Add(ctx, tenantID, "user-1"))
 
 	svc := tenantsvc.New(zap.NewNop(), tenants, providermemory.New(), bucketmemory.New(),
-		accesskeymemory.New(), delegationmemory.New(), secrets, wrapkeysmemory.New(), plcClient, nil)
-	return deleteEnv{svc: svc, tenants: tenants, directory: directory}
+		accesskeymemory.New(), principals, delegationmemory.New(), secrets, wrapkeysmemory.New(), plcClient, nil)
+	return deleteEnv{svc: svc, tenants: tenants, principals: principals, tenantID: tenantID, directory: directory}
 }
 
 func TestDelete(t *testing.T) {
@@ -261,6 +266,9 @@ func TestDelete(t *testing.T) {
 		require.NoError(t, env.svc.Delete(ctx, "tenant-1"))
 		_, err := env.tenants.GetByExternalID(ctx, "tenant-1")
 		require.ErrorIs(t, err, store.ErrRecordNotFound)
+		ps, err := env.principals.ListByTenant(ctx, env.tenantID)
+		require.NoError(t, err)
+		require.Empty(t, ps)
 		require.Equal(t, 1, env.directory.deactivations)
 	})
 

pkg/api/tenants_test.go

@@ -20,6 +20,7 @@ import (
 	accesskeymemory "github.com/fil-forge/hilt/pkg/store/accesskey/memory"
 	bucketmemory "github.com/fil-forge/hilt/pkg/store/bucket/memory"
 	delegationmemory "github.com/fil-forge/hilt/pkg/store/delegation/memory"
+	principalmemory "github.com/fil-forge/hilt/pkg/store/principal/memory"
 	"github.com/fil-forge/hilt/pkg/store/provider"
 	providermemory "github.com/fil-forge/hilt/pkg/store/provider/memory"
 	"github.com/fil-forge/hilt/pkg/store/tenant"
@@ -227,7 +228,7 @@ func setupProvision(t *testing.T, cfg *setupConfig) (*echo.Echo, *provisionDeps)
 		upload.WithHTTPClient(&http.Client{Transport: srv}))
 	require.NoError(t, err)
 
-	svc := tenantsvc.New(zap.NewNop(), deps.tenants, deps.providers, bucketmemory.New(), accesskeymemory.New(), delegationmemory.New(), deps.secrets, deps.wrapKeys, plcClient, upload)
+	svc := tenantsvc.New(zap.NewNop(), deps.tenants, deps.providers, bucketmemory.New(), accesskeymemory.New(), principalmemory.New(), delegationmemory.New(), deps.secrets, deps.wrapKeys, plcClient, upload)
 	route := api.NewProvisionTenantHandler(zap.NewNop(), svc)
 	e := echo.New()
 	e.Add(route.Method, route.Path, route.Handler)
@@ -468,7 +469,7 @@ func TestGetTenantHandler(t *testing.T) {
 	ctx := t.Context()
 	tenants := tenantmemory.New()
 	require.NoError(t, tenants.Add(ctx, testutil.RandomDID(t), "tenant-1", testutil.RandomDID(t), tenant.Active))
-	svc := tenantsvc.New(zap.NewNop(), tenants, providermemory.New(), bucketmemory.New(), accesskeymemory.New(), delegationmemory.New(), vaultmemory.New(), wrapkeymemory.New(), nil, nil)
+	svc := tenantsvc.New(zap.NewNop(), tenants, providermemory.New(), bucketmemory.New(), accesskeymemory.New(), principalmemory.New(), delegationmemory.New(), vaultmemory.New(), wrapkeymemory.New(), nil, nil)
 	e := serve(api.NewGetTenantHandler(zap.NewNop(), svc))
 
 	t.Run("found", func(t *testing.T) {
@@ -490,7 +491,7 @@ func TestUpdateTenantStatusHandler(t *testing.T) {
 	tenants := tenantmemory.New()
 	id := testutil.RandomDID(t)
 	require.NoError(t, tenants.Add(ctx, id, "tenant-1", testutil.RandomDID(t), tenant.Active))
-	svc := tenantsvc.New(zap.NewNop(), tenants, providermemory.New(), bucketmemory.New(), accesskeymemory.New(), delegationmemory.New(), vaultmemory.New(), wrapkeymemory.New(), nil, nil)
+	svc := tenantsvc.New(zap.NewNop(), tenants, providermemory.New(), bucketmemory.New(), accesskeymemory.New(), principalmemory.New(), delegationmemory.New(), vaultmemory.New(), wrapkeymemory.New(), nil, nil)
 	e := serve(api.NewUpdateTenantStatusHandler(zap.NewNop(), svc))
 
 	statusBody := func(s api.TenantStatus) []byte {
@@ -559,6 +560,7 @@ type deleteDeps struct {
 	tenants     *tenantmemory.Store
 	buckets     *bucketmemory.Store
 	accessKeys  *accesskeymemory.Store
+	principals  *principalmemory.Store
 	delegations *delegationmemory.Store
 	wrapKeys    *wrapkeymemory.Store
 	secrets     vault.Vault
@@ -613,6 +615,7 @@ func setupDelete(t *testing.T, status tenant.Status) (*echo.Echo, *deleteDeps) {
 		tenants:     tenantmemory.New(),
 		buckets:     bucketmemory.New(),
 		accessKeys:  accesskeymemory.New(),
+		principals:  principalmemory.New(),
 		delegations: delegationmemory.New(),
 		wrapKeys:    wrapkeymemory.New(),
 		secrets:     vaultmemory.New(),
@@ -636,7 +639,7 @@ func setupDelete(t *testing.T, status tenant.Status) (*echo.Echo, *deleteDeps) {
 		VaultKey: wrapkeystore.VaultKey(tenantID, 1),
 	}))
 
-	svc := tenantsvc.New(zap.NewNop(), deps.tenants, providermemory.New(), deps.buckets, deps.accessKeys, deps.delegations, deps.secrets, deps.wrapKeys, plcClient, nil)
+	svc := tenantsvc.New(zap.NewNop(), deps.tenants, providermemory.New(), deps.buckets, deps.accessKeys, deps.principals, deps.delegations, deps.secrets, deps.wrapKeys, plcClient, nil)
 	route := api.NewDeleteTenantHandler(zap.NewNop(), svc)
 	return serve(route), deps
 }
@@ -663,6 +666,7 @@ func TestDeleteTenantHandler(t *testing.T) {
 		require.NoError(t, deps.accessKeys.Add(ctx, akID, deps.tenantID, "k1", nil, []string{"s3:GetObject"}, nil))
 		akVaultKey := "/tenant/" + deps.tenantID.String() + "/access-key/" + akID.String()
 		require.NoError(t, deps.secrets.Write(ctx, akVaultKey, []byte("ak-key")))
+		require.NoError(t, deps.principals.Add(ctx, deps.tenantID, "user-1"))
 		require.NoError(t, deps.delegations.PutBatch(ctx, []ucan.Delegation{makeDelegation(t, deps.tenantID)}))
 		require.NoError(t, deps.delegations.PutBatch(ctx, []ucan.Delegation{makeDelegation(t, akID)}))
 
@@ -694,6 +698,11 @@ func TestDeleteTenantHandler(t *testing.T) {
 		_, err = deps.secrets.Read(ctx, akVaultKey)
 		require.ErrorIs(t, err, vault.ErrNotFound)
 
+		// Principals gone.
+		ps, err := deps.principals.ListByTenant(ctx, deps.tenantID)
+		require.NoError(t, err)
+		require.Empty(t, ps)
+
 		// Delegations to both the tenant and the access key gone.
 		tenantDlgs, err := deps.delegations.ListByAudience(ctx, deps.tenantID)
 		require.NoError(t, err)

pkg/fx/store/memory/provider.go

@@ -9,6 +9,8 @@ import (
 	membucket "github.com/fil-forge/hilt/pkg/store/bucket/memory"
 	"github.com/fil-forge/hilt/pkg/store/delegation"
 	memdelegation "github.com/fil-forge/hilt/pkg/store/delegation/memory"
+	"github.com/fil-forge/hilt/pkg/store/principal"
+	memprincipal "github.com/fil-forge/hilt/pkg/store/principal/memory"
 	"github.com/fil-forge/hilt/pkg/store/provider"
 	memprovider "github.com/fil-forge/hilt/pkg/store/provider/memory"
 	"github.com/fil-forge/hilt/pkg/store/tenant"
@@ -24,6 +26,7 @@ var Module = fx.Module("memory-store",
 		fx.Annotate(memaccesskey.New, fx.As(new(accesskey.Store))),
 		fx.Annotate(membucket.New, fx.As(new(bucket.Store))),
 		fx.Annotate(memdelegation.New, fx.As(new(delegation.Store))),
+		fx.Annotate(memprincipal.New, fx.As(new(principal.Store))),
 		fx.Annotate(memprovider.New, fx.As(new(provider.Store))),
 		fx.Annotate(memtenant.New, fx.As(new(tenant.Store))),
 		fx.Annotate(memwrapkey.New, fx.As(new(wrapkey.Store))),

pkg/fx/store/postgres/provider.go

@@ -16,6 +16,8 @@ import (
 	pgbucket "github.com/fil-forge/hilt/pkg/store/bucket/postgres"
 	"github.com/fil-forge/hilt/pkg/store/delegation"
 	pgdelegation "github.com/fil-forge/hilt/pkg/store/delegation/postgres"
+	"github.com/fil-forge/hilt/pkg/store/principal"
+	pgprincipal "github.com/fil-forge/hilt/pkg/store/principal/postgres"
 	"github.com/fil-forge/hilt/pkg/store/provider"
 	pgprovider "github.com/fil-forge/hilt/pkg/store/provider/postgres"
 	"github.com/fil-forge/hilt/pkg/store/tenant"
@@ -35,6 +37,7 @@ var Module = fx.Module("postgres-store",
 		NewAccessKeyStore,
 		NewBucketStore,
 		NewDelegationStore,
+		NewPrincipalStore,
 		NewProviderStore,
 		NewTenantStore,
 		NewWrapKeyStore,
@@ -120,6 +123,10 @@ func NewDelegationStore(mdb *MigratedPool) delegation.Store {
 	return pgdelegation.New(mdb.Pool)
 }
 
+func NewPrincipalStore(mdb *MigratedPool) principal.Store {
+	return pgprincipal.New(mdb.Pool)
+}
+
 func NewProviderStore(mdb *MigratedPool) provider.Store {
 	return pgprovider.New(mdb.Pool)
 }

pkg/migrations/sql/00006_principal.sql

@@ -0,0 +1,20 @@
+-- +goose Up
+-- +goose StatementBegin
+-- principal is a console user within a tenant, identified by the console's
+-- principalId. A principal holds no key material and no delegations: its access
+-- is computed from bucket policies at request time. Removal sets deleted_at
+-- rather than deleting the row, so a later PUT for the same id revives it and
+-- an id reused by the console never inherits anything.
+CREATE TABLE principal (
+    tenant_id   TEXT        NOT NULL REFERENCES tenant(id) ON DELETE RESTRICT, -- DID (did:plc)
+    external_id TEXT        NOT NULL,                                          -- console principalId
+    created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    deleted_at  TIMESTAMPTZ,                                                   -- set by removal; cleared by revive
+    PRIMARY KEY (tenant_id, external_id)
+);
+-- +goose StatementEnd
+
+-- +goose Down
+-- +goose StatementBegin
+DROP TABLE IF EXISTS principal;
+-- +goose StatementEnd