KYC Onboarding Backend: Document Uploads, Verification and Webhooks

-- onboarding/migrations/1_create_applications.up.sql
CREATE TABLE applications (
    id BIGSERIAL PRIMARY KEY,
    customer_id TEXT NOT NULL UNIQUE,
    status TEXT NOT NULL DEFAULT 'collecting', -- collecting, submitted, approved, rejected, needs_review
    submitted_event_at TIMESTAMPTZ,
    provider_check_id TEXT UNIQUE,
    reason TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    decided_at TIMESTAMPTZ
);

CREATE TABLE application_events (
    application_id BIGINT PRIMARY KEY REFERENCES applications (id),
    customer_id    TEXT NOT NULL,
    published_at   TIMESTAMPTZ
);

// onboarding/onboarding.go
package onboarding

import (
	"context"
	"fmt"
	"time"

	"encore.dev/beta/errs"
	"encore.dev/cron"
	"encore.dev/pubsub"
	"encore.dev/storage/objects"
	"encore.dev/storage/sqldb"
)

var db = sqldb.NewDatabase("onboarding", sqldb.DatabaseConfig{Migrations: "./migrations"})

// Private bucket for identity documents. Clients upload straight to it.
var Documents = objects.NewBucket("kyc-documents", objects.BucketConfig{})

type Submitted struct {
	ApplicationID int64
	CustomerID    string
}

var ApplicationSubmitted = pubsub.NewTopic[*Submitted]("kyc-submitted", pubsub.TopicConfig{
	DeliveryGuarantee: pubsub.AtLeastOnce,
})

type StartResponse struct {
	ApplicationID int64
	UploadURLs    map[string]string // document kind -> signed PUT URL
}

// Start opens an application and returns short-lived upload URLs for the documents.
//
//encore:api private method=POST path=/onboarding/customers/:customerID/start
func Start(ctx context.Context, customerID string) (*StartResponse, error) {
	var id int64
	err := db.QueryRow(ctx, `
		INSERT INTO applications (customer_id) VALUES ($1)
		ON CONFLICT (customer_id) DO UPDATE SET customer_id = EXCLUDED.customer_id
		RETURNING id
	`, customerID).Scan(&id)
	if err != nil {
		return nil, err
	}
	urls := map[string]string{}
	for _, kind := range []string{"id-front", "id-back", "selfie"} {
		key := fmt.Sprintf("%d/%s", id, kind)
		u, err := Documents.SignedUploadURL(ctx, key, objects.WithTTL(15*time.Minute))
		if err != nil {
			return nil, err
		}
		urls[kind] = u.URL
	}
	return &StartResponse{ApplicationID: id, UploadURLs: urls}, nil
}

const maxDocumentBytes = 10 << 20 // 10 MiB

func allowedDocumentType(ct string) bool {
	return ct == "image/jpeg" || ct == "image/png" || ct == "application/pdf"
}

func acceptDocument(ctx context.Context, key, kind string) error {
	attrs, err := Documents.Attrs(ctx, key)
	if err != nil {
		return &errs.Error{Code: errs.FailedPrecondition, Message: kind + " not uploaded"}
	}
	if attrs.Size > maxDocumentBytes || !allowedDocumentType(attrs.ContentType) {
		_ = Documents.Remove(ctx, key)
		return &errs.Error{Code: errs.InvalidArgument, Message: kind + " must be a JPEG, PNG or PDF up to 10 MiB"}
	}
	return nil
}

// Submit checks that every document arrived and is a valid identity file, then
// hands off to the verification worker.
//
//encore:api private method=POST path=/onboarding/applications/:id/submit
func Submit(ctx context.Context, id int64) error {
	for _, kind := range []string{"id-front", "id-back", "selfie"} {
		if err := acceptDocument(ctx, fmt.Sprintf("%d/%s", id, kind), kind); err != nil {
			return err
		}
	}
	tx, err := db.Begin(ctx)
	if err != nil {
		return err
	}
	defer tx.Rollback()

	var customerID string
	err = tx.QueryRow(ctx, `
		UPDATE applications SET status = 'submitted'
		WHERE id = $1 AND status IN ('collecting', 'submitted')
		RETURNING customer_id
	`, id).Scan(&customerID)
	if err == sqldb.ErrNoRows {
		return nil
	} else if err != nil {
		return err
	}
	if _, err = tx.Exec(ctx, `
		INSERT INTO application_events (application_id, customer_id)
		VALUES ($1, $2)
		ON CONFLICT (application_id) DO NOTHING
	`, id, customerID); err != nil {
		return err
	}
	if err := tx.Commit(); err != nil {
		return err
	}
	return publishApplicationEvent(ctx, id)
}

var _ = cron.NewJob("recover-kyc-events", cron.JobConfig{
	Title:    "Publish unpublished KYC submission events",
	Every:    1 * cron.Hour,
	Endpoint: RecoverSubmittedEvents,
})

// RecoverSubmittedEvents is called by the Encore platform on a schedule,
// and from Submit, so a crash after the outbox write is retried even if
// the client never comes back. Cron jobs do not run locally or in preview
// environments; call this endpoint from the dashboard to test it.
//
//encore:api private
func RecoverSubmittedEvents(ctx context.Context) error {
	return publishPendingApplicationEvents(ctx)
}

func publishApplicationEvent(ctx context.Context, id int64) error {
	var customerID string
	err := db.QueryRow(ctx, `
		SELECT customer_id FROM application_events
		WHERE application_id = $1 AND published_at IS NULL
	`, id).Scan(&customerID)
	if err == sqldb.ErrNoRows {
		return nil
	} else if err != nil {
		return err
	}
	if _, err := ApplicationSubmitted.Publish(ctx, &Submitted{ApplicationID: id, CustomerID: customerID}); err != nil {
		return err
	}
	if _, err := db.Exec(ctx, `
		UPDATE application_events SET published_at = now()
		WHERE application_id = $1 AND published_at IS NULL
	`, id); err != nil {
		return err
	}
	_, err = db.Exec(ctx, `
		UPDATE applications SET submitted_event_at = now()
		WHERE id = $1 AND submitted_event_at IS NULL
	`, id)
	return err
}

func publishPendingApplicationEvents(ctx context.Context) error {
	rows, err := db.Query(ctx, `
		SELECT application_id
		FROM application_events
		WHERE published_at IS NULL
	`)
	if err != nil {
		return err
	}
	defer rows.Close()
	for rows.Next() {
		var id int64
		if err := rows.Scan(&id); err != nil {
			return err
		}
		if err := publishApplicationEvent(ctx, id); err != nil {
			return err
		}
	}
	return rows.Err()
}

// onboarding/verify.go
package onboarding

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"io"
	"net/http"

	"encore.dev/pubsub"
)

var secrets struct {
	VerifyAPIKey        string
	VerifyWebhookSecret string
}

// startCheck (omitted here) builds signed download URLs, calls the provider
// with Idempotency-Key kyc-<application id>, and stores the returned check ID.

var _ = pubsub.NewSubscription(ApplicationSubmitted, "start-check", pubsub.SubscriptionConfig[*Submitted]{
	Handler:     startCheck,
	AckDeadline: 2 * time.Minute,
	RetryPolicy: &pubsub.RetryPolicy{MinBackoff: 30 * time.Second, MaxRetries: 10},
})

//encore:api public raw method=POST path=/webhooks/verification
func VerificationResult(w http.ResponseWriter, req *http.Request) {
	body, err := io.ReadAll(io.LimitReader(req.Body, 1<<20))
	if err != nil {
		http.Error(w, "bad body", http.StatusBadRequest)
		return
	}
	mac := hmac.New(sha256.New, []byte(secrets.VerifyWebhookSecret))
	mac.Write(body)
	if !hmac.Equal([]byte(hex.EncodeToString(mac.Sum(nil))), []byte(req.Header.Get("X-Signature"))) {
		http.Error(w, "bad signature", http.StatusUnauthorized)
		return
	}
	var ev struct {
		CheckID string `json:"check_id"`
		Outcome string `json:"outcome"` // approved, rejected, needs_review
		Reason  string `json:"reason"`
	}
	if err := json.Unmarshal(body, &ev); err != nil {
		http.Error(w, "bad json", http.StatusBadRequest)
		return
	}
	// Only the first decision counts; duplicates and late retries change nothing.
	if _, err = db.Exec(req.Context(), `
		UPDATE applications SET status = $1, reason = $2, decided_at = now()
		WHERE provider_check_id = $3 AND status = 'submitted'
	`, ev.Outcome, ev.Reason, ev.CheckID); err != nil {
		http.Error(w, "retry later", http.StatusInternalServerError)
		return
	}
	w.WriteHeader(http.StatusNoContent)
}