feat: implement customer add handler

go.mod

@@ -11,7 +11,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/dynamodb v1.57.3
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.96.4
 	github.com/docker/docker v28.5.2+incompatible
-	github.com/fil-forge/libforge v0.0.0-20260630210927-2b55dbcf944f
+	github.com/fil-forge/libforge v0.0.0-20260701162346-f0706e1641a3
 	github.com/fil-forge/ucantone v0.0.0-20260619013642-7985ec010b88
 	github.com/google/uuid v1.6.0
 	github.com/ipfs/go-cid v0.6.1

go.sum

@@ -92,8 +92,8 @@ github.com/ebitengine/purego v0.10.0 h1:QIw4xfpWT6GWTzaW5XEKy3HXoqrJGx1ijYHzTF0/
 github.com/ebitengine/purego v0.10.0/go.mod h1:iIjxzd6CiRiOG0UyXP+V1+jWqUXVjPKLAI0mRfJZTmQ=
 github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg=
 github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
-github.com/fil-forge/libforge v0.0.0-20260630210927-2b55dbcf944f h1:R73fE6WIAJF0tA7YGxJx7rxslPbrGxGnqkh8HOlWyc0=
-github.com/fil-forge/libforge v0.0.0-20260630210927-2b55dbcf944f/go.mod h1:0kXihIQ4L2uZ00nR5XrZ/Y8Db7Ht/qQNuiWslwMJ95M=
+github.com/fil-forge/libforge v0.0.0-20260701162346-f0706e1641a3 h1:/EDxpbuVeSXH3FLF7MK4G4wrDUFmCO9fLQHL23RZ2uU=
+github.com/fil-forge/libforge v0.0.0-20260701162346-f0706e1641a3/go.mod h1:0kXihIQ4L2uZ00nR5XrZ/Y8Db7Ht/qQNuiWslwMJ95M=
 github.com/fil-forge/ucantone v0.0.0-20260619013642-7985ec010b88 h1:N0gbL3Ik+XBYk4y/5BxTVymwbRGlxRXwC5eNWzi1bGI=
 github.com/fil-forge/ucantone v0.0.0-20260619013642-7985ec010b88/go.mod h1:rTIRXz4xErI4U+YlBU9ZvhlTbr4Hs5tJhVMwereVkSg=
 github.com/filecoin-project/go-data-segment v0.0.1 h1:1wmDxOG4ubWQm3ZC1XI5nCon5qgSq7Ra3Rb6Dbu10Gs=

internal/fx/service/handlers/provider.go

@@ -24,6 +24,10 @@ var Module = fx.Module("service-handlers",
 			handlers.NewAccessRequestHandler,
 			fx.ResultTags(`group:"ucan_handlers"`),
 		),
+		fx.Annotate(
+			handlers.NewCustomerAddHandler,
+			fx.ResultTags(`group:"ucan_handlers"`),
+		),
 		fx.Annotate(
 			handlers.NewAdminProviderDeregisterHandler,
 			fx.ResultTags(`group:"ucan_handlers"`),

internal/migrations/sql/00001_init.sql

@@ -2,15 +2,15 @@
 -- +goose StatementBegin
 CREATE TABLE customer (
     customer          TEXT        PRIMARY KEY,
-    account           TEXT,
+    external_account  TEXT,
     product           TEXT        NOT NULL,
     details           JSONB,
     reserved_capacity BIGINT,
     inserted_at       TIMESTAMPTZ NOT NULL,
     updated_at        TIMESTAMPTZ
 );
 
-CREATE INDEX customer_account_idx ON customer (account) WHERE account IS NOT NULL;
+CREATE INDEX customer_external_account_idx ON customer (external_account) WHERE external_account IS NOT NULL;
 
 CREATE TABLE storage_provider (
     provider           TEXT        PRIMARY KEY,

pkg/service/handlers/customer_add.go

@@ -0,0 +1,65 @@
+package handlers
+
+import (
+	"fmt"
+
+	customercmds "github.com/fil-forge/libforge/commands/customer"
+	"github.com/fil-forge/libforge/identity"
+	customerstore "github.com/fil-forge/sprue/pkg/store/customer"
+	"github.com/fil-forge/ucantone/binding"
+	"github.com/fil-forge/ucantone/errors"
+	"github.com/fil-forge/ucantone/server"
+	"go.uber.org/zap"
+)
+
+// InvalidCustomerSubjectErrorName is the error name returned when a
+// /customer/add invocation's subject is not the service DID.
+const InvalidCustomerSubjectErrorName = "InvalidCustomerSubject"
+
+// errInvalidCustomerSubject is returned when the invocation subject is not the
+// service DID.
+var errInvalidCustomerSubject = errors.New(InvalidCustomerSubjectErrorName, "invocation subject must be the service")
+
+// NewCustomerAddHandler handles /customer/add invocations. It asserts that the
+// invocation subject is the service DID and registers the customer in the
+// customer store.
+func NewCustomerAddHandler(id identity.Identity, customerStore customerstore.Store, logger *zap.Logger) server.Route {
+	log := logger.With(zap.Stringer("handler", customercmds.Add))
+	return customercmds.Add.Route(
+		func(req *binding.Request[*customercmds.AddArguments], res *binding.Response[*customercmds.AddOK]) error {
+			if req.Invocation().Subject() != id.DID() {
+				log.Warn("not a valid invocation", zap.Stringer("subject", req.Invocation().Subject()))
+				return res.SetFailure(errInvalidCustomerSubject)
+			}
+
+			args := req.Task().Arguments()
+
+			log := log.With(
+				zap.Stringer("customer", args.Customer),
+				zap.Stringer("product", args.Product),
+			)
+			log.Debug("adding customer")
+
+			var details map[string]any
+			if len(args.Details) > 0 {
+				details = make(map[string]any, len(args.Details))
+				for k, v := range args.Details {
+					details[k] = v
+				}
+			}
+
+			err := customerStore.Add(req.Context(), args.Customer, args.ExternalAccount, args.Product, details, nil)
+			if err != nil {
+				if errors.Is(err, customerstore.ErrCustomerExists) {
+					log.Warn("customer already exists")
+					return res.SetFailure(err)
+				}
+				log.Error("failed to add customer", zap.Error(err))
+				return fmt.Errorf("adding customer: %w", err)
+			}
+
+			log.Debug("customer added successfully")
+			return res.SetSuccess(&customercmds.AddOK{})
+		},
+	)
+}

pkg/service/handlers/customer_add_test.go

@@ -0,0 +1,123 @@
+package handlers
+
+import (
+	"context"
+	"testing"
+
+	customercmds "github.com/fil-forge/libforge/commands/customer"
+	"github.com/fil-forge/libforge/identity"
+	"github.com/fil-forge/sprue/internal/testutil"
+	customerstore "github.com/fil-forge/sprue/pkg/store/customer"
+	customer_store "github.com/fil-forge/sprue/pkg/store/customer/memory"
+	"github.com/fil-forge/ucantone/did"
+	"github.com/fil-forge/ucantone/execution"
+	"github.com/fil-forge/ucantone/ucan/invocation"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap/zaptest"
+)
+
+// invokeCustomerAdd builds a /customer/add invocation with the given subject,
+// plus a signed response, ready to pass to the handler.
+func invokeCustomerAdd(
+	t *testing.T,
+	ctx context.Context,
+	id identity.Identity,
+	subject did.DID,
+	args *customercmds.AddArguments,
+) (execution.Request, *execution.ExecResponse) {
+	t.Helper()
+	inv, err := customercmds.Add.Invoke(
+		id.Issuer,
+		subject,
+		args,
+		invocation.WithAudience(id.Issuer.DID()),
+	)
+	require.NoError(t, err)
+	req := execution.NewRequest(ctx, inv)
+	res, err := execution.NewResponse(req.Invocation().Task().Link(), execution.WithIssuer(id.Issuer))
+	require.NoError(t, err)
+	return req, res
+}
+
+func TestCustomerAddHandler(t *testing.T) {
+	logger := zaptest.NewLogger(t)
+	ctx := t.Context()
+	id := newTestIdentity(t)
+
+	t.Run("success", func(t *testing.T) {
+		store := customer_store.New()
+		handler := NewCustomerAddHandler(id, store, logger)
+
+		customerDID := testutil.RandomDID(t)
+		product := testutil.Must(did.Parse("did:web:free.web3.storage"))(t)
+		account := "stripe:cus_9s6XKzkNRiz8i3"
+		details := map[string]string{"plan": "pro"}
+
+		// Subject is the service DID, as set by the upload client.
+		req, res := invokeCustomerAdd(t, ctx, id, id.DID(), &customercmds.AddArguments{
+			Customer:        customerDID,
+			ExternalAccount: &account,
+			Product:         product,
+			Details:         details,
+		})
+
+		err := handler.Handler(req, res)
+		require.NoError(t, err)
+
+		_, err = customercmds.Add.Unpack(res.Receipt())
+		require.NoError(t, err)
+
+		rec, err := store.Get(ctx, customerDID)
+		require.NoError(t, err)
+		require.Equal(t, customerDID, rec.Customer)
+		require.Equal(t, product, rec.Product)
+		require.NotNil(t, rec.ExternalAccount)
+		require.Equal(t, account, *rec.ExternalAccount)
+		require.Equal(t, map[string]any{"plan": "pro"}, rec.Details)
+	})
+
+	t.Run("wrong subject", func(t *testing.T) {
+		store := customer_store.New()
+		handler := NewCustomerAddHandler(id, store, logger)
+
+		customerDID := testutil.RandomDID(t)
+		product := testutil.Must(did.Parse("did:web:free.web3.storage"))(t)
+		notService := testutil.RandomIssuer(t)
+
+		req, res := invokeCustomerAdd(t, ctx, id, notService.DID(), &customercmds.AddArguments{
+			Customer: customerDID,
+			Product:  product,
+		})
+
+		err := handler.Handler(req, res)
+		require.NoError(t, err)
+
+		_, err = customercmds.Add.Unpack(res.Receipt())
+		require.ErrorIs(t, err, errInvalidCustomerSubject)
+
+		// The customer must not have been written.
+		_, err = store.Get(ctx, customerDID)
+		require.ErrorIs(t, err, customerstore.ErrCustomerNotFound)
+	})
+
+	t.Run("duplicate customer", func(t *testing.T) {
+		store := customer_store.New()
+		handler := NewCustomerAddHandler(id, store, logger)
+
+		customerDID := testutil.RandomDID(t)
+		product := testutil.Must(did.Parse("did:web:free.web3.storage"))(t)
+
+		require.NoError(t, store.Add(ctx, customerDID, nil, product, nil, nil))
+
+		req, res := invokeCustomerAdd(t, ctx, id, id.DID(), &customercmds.AddArguments{
+			Customer: customerDID,
+			Product:  product,
+		})
+
+		err := handler.Handler(req, res)
+		require.NoError(t, err)
+
+		_, err = customercmds.Add.Unpack(res.Receipt())
+		require.ErrorIs(t, err, customerstore.ErrCustomerExists)
+	})
+}

pkg/store/customer/aws/store.go

@@ -33,16 +33,16 @@ var DynamoCustomerTableProps = struct {
 			AttributeType: types.ScalarAttributeTypeS,
 		},
 		{
-			AttributeName: aws.String("account"),
+			AttributeName: aws.String("externalAccount"),
 			AttributeType: types.ScalarAttributeTypeS,
 		},
 	},
 	GSI: []types.GlobalSecondaryIndex{
 		{
-			IndexName: aws.String("account"),
+			IndexName: aws.String("externalAccount"),
 			KeySchema: []types.KeySchemaElement{
 				{
-					AttributeName: aws.String("account"),
+					AttributeName: aws.String("externalAccount"),
 					KeyType:       types.KeyTypeHash,
 				},
 			},
@@ -235,8 +235,8 @@ func itemToRecord(item map[string]types.AttributeValue) (customer.Record, error)
 		Product:  product,
 	}
 
-	if accountAttr, ok := item["account"].(*types.AttributeValueMemberS); ok {
-		rec.Account = &accountAttr.Value
+	if externalAccAttr, ok := item["externalAccount"].(*types.AttributeValueMemberS); ok {
+		rec.ExternalAccount = &externalAccAttr.Value
 	}
 
 	if capAttr, ok := item["reservedCapacity"].(*types.AttributeValueMemberN); ok {

pkg/store/customer/customer.go

@@ -40,7 +40,7 @@ type Record struct {
 	Customer did.DID
 	// Opaque identifier representing an account in the payment system
 	// e.g. Stripe customer ID (stripe:cus_9s6XKzkNRiz8i3)
-	Account *string
+	ExternalAccount *string
 	// Unique identifier of the product a.k.a plan.
 	Product did.DID
 	// Misc customer details

pkg/store/customer/customer_test.go

@@ -85,7 +85,7 @@ func TestCustomerStore(t *testing.T) {
 				require.NoError(t, err)
 				require.Equal(t, customerID, rec.Customer)
 				require.Equal(t, product, rec.Product)
-				require.Nil(t, rec.Account)
+				require.Nil(t, rec.ExternalAccount)
 				require.False(t, rec.InsertedAt.IsZero())
 			})
 
@@ -99,7 +99,7 @@ func TestCustomerStore(t *testing.T) {
 
 				rec, err := s.Get(t.Context(), customerID)
 				require.NoError(t, err)
-				require.Equal(t, &account, rec.Account)
+				require.Equal(t, &account, rec.ExternalAccount)
 				require.Equal(t, &capacity, rec.ReservedCapacity)
 			})
 

pkg/store/customer/memory/memory.go

@@ -67,7 +67,7 @@ func (s *Store) List(ctx context.Context, options ...customer.ListOption) (store
 	}, nil
 }
 
-func (s *Store) Add(ctx context.Context, customerID did.DID, account *string, product did.DID, details map[string]any, reservedCapacity *uint64) error {
+func (s *Store) Add(ctx context.Context, customerID did.DID, externalAccount *string, product did.DID, details map[string]any, reservedCapacity *uint64) error {
 	s.mutex.Lock()
 	defer s.mutex.Unlock()
 
@@ -78,7 +78,7 @@ func (s *Store) Add(ctx context.Context, customerID did.DID, account *string, pr
 	}
 	c := customer.Record{
 		Customer:         customerID,
-		Account:          account,
+		ExternalAccount:  externalAccount,
 		Product:          product,
 		Details:          details,
 		ReservedCapacity: reservedCapacity,

pkg/store/customer/postgres/store.go

@@ -38,7 +38,7 @@ func (s *Store) Initialize(ctx context.Context) error { return nil }
 
 func (s *Store) Get(ctx context.Context, customerID did.DID) (customer.Record, error) {
 	row := s.pool.QueryRow(ctx, `
-		SELECT customer, account, product, details, reserved_capacity, inserted_at, updated_at
+		SELECT customer, external_account, product, details, reserved_capacity, inserted_at, updated_at
 		FROM customer
 		WHERE customer = $1
 	`, customerID.String())
@@ -52,7 +52,7 @@ func (s *Store) Get(ctx context.Context, customerID did.DID) (customer.Record, e
 	return rec, nil
 }
 
-func (s *Store) Add(ctx context.Context, customerID did.DID, account *string, product did.DID, details map[string]any, reservedCapacity *uint64) error {
+func (s *Store) Add(ctx context.Context, customerID did.DID, externalAccount *string, product did.DID, details map[string]any, reservedCapacity *uint64) error {
 	var detailsJSON []byte
 	if len(details) > 0 {
 		b, err := json.Marshal(details)
@@ -69,9 +69,9 @@ func (s *Store) Add(ctx context.Context, customerID did.DID, account *string, pr
 	}
 
 	_, err := s.pool.Exec(ctx, `
-		INSERT INTO customer (customer, account, product, details, reserved_capacity, inserted_at)
+		INSERT INTO customer (customer, external_account, product, details, reserved_capacity, inserted_at)
 		VALUES ($1, $2, $3, $4, $5, $6)
-	`, customerID.String(), account, product.String(), detailsJSON, capacity, time.Now().UTC())
+	`, customerID.String(), externalAccount, product.String(), detailsJSON, capacity, time.Now().UTC())
 
 	if err != nil {
 		var pgErr *pgconn.PgError
@@ -96,7 +96,7 @@ func (s *Store) List(ctx context.Context, options ...customer.ListOption) (store
 
 	args := []any{limit + 1}
 	query := `
-		SELECT customer, account, product, details, reserved_capacity, inserted_at, updated_at
+		SELECT customer, external_account, product, details, reserved_capacity, inserted_at, updated_at
 		FROM customer
 	`
 	if cfg.Cursor != nil {
@@ -154,15 +154,15 @@ type rowScanner interface {
 
 func scanRecord(row rowScanner) (customer.Record, error) {
 	var (
-		customerStr  string
-		account      *string
-		productStr   string
-		detailsJSON  []byte
-		capacity     *int64
-		insertedAt   time.Time
-		updatedAtRaw *time.Time
+		customerStr     string
+		externalAccount *string
+		productStr      string
+		detailsJSON     []byte
+		capacity        *int64
+		insertedAt      time.Time
+		updatedAtRaw    *time.Time
 	)
-	if err := row.Scan(&customerStr, &account, &productStr, &detailsJSON, &capacity, &insertedAt, &updatedAtRaw); err != nil {
+	if err := row.Scan(&customerStr, &externalAccount, &productStr, &detailsJSON, &capacity, &insertedAt, &updatedAtRaw); err != nil {
 		return customer.Record{}, err
 	}
 
@@ -176,10 +176,10 @@ func scanRecord(row rowScanner) (customer.Record, error) {
 	}
 
 	rec := customer.Record{
-		Customer:   customerID,
-		Account:    account,
-		Product:    product,
-		InsertedAt: insertedAt,
+		Customer:        customerID,
+		ExternalAccount: externalAccount,
+		Product:         product,
+		InsertedAt:      insertedAt,
 	}
 	if updatedAtRaw != nil {
 		rec.UpdatedAt = *updatedAtRaw