mirror of
https://github.com/luxfi/broker.git
synced 2026-09-24 00:25:18 +00:00
feat: auto-provision Alpaca account on application approval
This commit is contained in:
+1
-1
@@ -89,7 +89,7 @@ func main() {
|
||||
log.Info().Msg("Compliance store seeded with demo data")
|
||||
}
|
||||
scamDB := compliance.NewScamDB()
|
||||
srv.Mount("/compliance", compliance.NewRouter(complianceStore, compliance.WithScamDB(scamDB)))
|
||||
srv.Mount("/compliance", compliance.NewRouter(complianceStore, compliance.WithScamDB(scamDB), compliance.WithRegistry(registry)))
|
||||
log.Info().Msg("Compliance routes mounted at /compliance")
|
||||
|
||||
// --- gRPC Server (optional) ---
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package compliance
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/hmac"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
@@ -10,6 +11,8 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
"github.com/luxfi/broker/pkg/provider"
|
||||
"github.com/luxfi/broker/pkg/types"
|
||||
"github.com/luxfi/compliance/pkg/onboarding"
|
||||
"github.com/rs/zerolog/log"
|
||||
)
|
||||
@@ -33,7 +36,8 @@ func init() {
|
||||
|
||||
// applicationHandler manages the 5-step investor onboarding flow.
|
||||
type applicationHandler struct {
|
||||
store ComplianceStore
|
||||
store ComplianceStore
|
||||
registry *provider.Registry
|
||||
}
|
||||
|
||||
// newApplicationSteps delegates to the compliance library's canonical step definitions.
|
||||
@@ -483,9 +487,68 @@ func (h *applicationHandler) handleReview(w http.ResponseWriter, r *http.Request
|
||||
writeError(w, http.StatusInternalServerError, "internal error")
|
||||
return
|
||||
}
|
||||
|
||||
// On approval, provision Alpaca brokerage account asynchronously.
|
||||
if app.Status == AppApproved && h.registry != nil {
|
||||
go h.provisionAlpacaAccount(app)
|
||||
}
|
||||
|
||||
writeJSON(w, http.StatusOK, app)
|
||||
}
|
||||
|
||||
// provisionAlpacaAccount creates an Alpaca brokerage account for an approved application.
|
||||
func (h *applicationHandler) provisionAlpacaAccount(app *Application) {
|
||||
alpaca, err := h.registry.Get("alpaca")
|
||||
if err != nil {
|
||||
log.Error().Err(err).Str("app", app.ID).Msg("no alpaca provider for provisioning")
|
||||
return
|
||||
}
|
||||
|
||||
street := []string{}
|
||||
if app.AddressLine1 != "" {
|
||||
street = append(street, app.AddressLine1)
|
||||
}
|
||||
if app.AddressLine2 != "" {
|
||||
street = append(street, app.AddressLine2)
|
||||
}
|
||||
|
||||
country := app.Country
|
||||
if country == "" {
|
||||
country = "USA"
|
||||
}
|
||||
|
||||
acctReq := &types.CreateAccountRequest{
|
||||
Identity: &types.Identity{
|
||||
GivenName: app.FirstName,
|
||||
FamilyName: app.LastName,
|
||||
DateOfBirth: app.DateOfBirth,
|
||||
CountryOfTax: country,
|
||||
},
|
||||
Contact: &types.Contact{
|
||||
Email: app.Email,
|
||||
Phone: app.Phone,
|
||||
Street: street,
|
||||
City: app.City,
|
||||
State: app.State,
|
||||
PostalCode: app.ZipCode,
|
||||
Country: country,
|
||||
},
|
||||
}
|
||||
|
||||
acct, err := alpaca.CreateAccount(context.Background(), acctReq)
|
||||
if err != nil {
|
||||
log.Error().Err(err).Str("app", app.ID).Msg("alpaca account creation failed")
|
||||
return
|
||||
}
|
||||
|
||||
app.AlpacaAccountID = acct.ID
|
||||
if err := h.store.SaveApplication(app); err != nil {
|
||||
log.Error().Err(err).Str("app", app.ID).Str("alpaca_id", acct.ID).Msg("failed to save alpaca account ID")
|
||||
return
|
||||
}
|
||||
log.Info().Str("app", app.ID).Str("alpaca_id", acct.ID).Msg("alpaca account provisioned")
|
||||
}
|
||||
|
||||
// handleGetDocuments lists documents for an application.
|
||||
func (h *applicationHandler) handleGetDocuments(w http.ResponseWriter, r *http.Request) {
|
||||
id := chi.URLParam(r, "id")
|
||||
|
||||
@@ -2,12 +2,16 @@ package compliance
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
"github.com/luxfi/broker/pkg/provider"
|
||||
"github.com/luxfi/broker/pkg/types"
|
||||
)
|
||||
|
||||
func newTestRouter() (chi.Router, *MemoryStore) {
|
||||
@@ -3369,3 +3373,177 @@ func TestScamDBCount(t *testing.T) {
|
||||
t.Fatalf("expected at least 100 default addresses, got %d", db.Count())
|
||||
}
|
||||
}
|
||||
|
||||
// ==========================================================================
|
||||
// Alpaca Provisioning on Approval
|
||||
// ==========================================================================
|
||||
|
||||
// mockAlpacaProvider implements provider.Provider for testing account provisioning.
|
||||
type mockAlpacaProvider struct {
|
||||
created chan *types.CreateAccountRequest
|
||||
}
|
||||
|
||||
func (m *mockAlpacaProvider) Name() string { return "alpaca" }
|
||||
func (m *mockAlpacaProvider) CreateAccount(_ context.Context, req *types.CreateAccountRequest) (*types.Account, error) {
|
||||
m.created <- req
|
||||
return &types.Account{ID: "mock-alpaca-acct-001", Provider: "alpaca"}, nil
|
||||
}
|
||||
func (m *mockAlpacaProvider) GetAccount(context.Context, string) (*types.Account, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockAlpacaProvider) ListAccounts(context.Context) ([]*types.Account, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockAlpacaProvider) GetPortfolio(context.Context, string) (*types.Portfolio, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockAlpacaProvider) CreateOrder(context.Context, string, *types.CreateOrderRequest) (*types.Order, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockAlpacaProvider) ListOrders(context.Context, string) ([]*types.Order, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockAlpacaProvider) GetOrder(context.Context, string, string) (*types.Order, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockAlpacaProvider) CancelOrder(context.Context, string, string) error { return nil }
|
||||
func (m *mockAlpacaProvider) CreateTransfer(context.Context, string, *types.CreateTransferRequest) (*types.Transfer, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockAlpacaProvider) ListTransfers(context.Context, string) ([]*types.Transfer, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockAlpacaProvider) CreateBankRelationship(context.Context, string, string, string, string, string) (*types.BankRelationship, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockAlpacaProvider) ListBankRelationships(context.Context, string) ([]*types.BankRelationship, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockAlpacaProvider) ListAssets(context.Context, string) ([]*types.Asset, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockAlpacaProvider) GetAsset(context.Context, string) (*types.Asset, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockAlpacaProvider) GetSnapshot(context.Context, string) (*types.MarketSnapshot, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockAlpacaProvider) GetSnapshots(context.Context, []string) (map[string]*types.MarketSnapshot, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockAlpacaProvider) GetBars(context.Context, string, string, string, string, int) ([]*types.Bar, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockAlpacaProvider) GetLatestTrades(context.Context, []string) (map[string]*types.Trade, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockAlpacaProvider) GetLatestQuotes(context.Context, []string) (map[string]*types.Quote, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockAlpacaProvider) GetClock(context.Context) (*types.MarketClock, error) {
|
||||
return nil, nil
|
||||
}
|
||||
func (m *mockAlpacaProvider) GetCalendar(context.Context, string, string) ([]*types.MarketCalendarDay, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func TestApplicationApprovalProvisionsAlpacaAccount(t *testing.T) {
|
||||
mock := &mockAlpacaProvider{created: make(chan *types.CreateAccountRequest, 1)}
|
||||
reg := provider.NewRegistry()
|
||||
reg.Register(mock)
|
||||
|
||||
store := NewStore()
|
||||
router := NewRouter(store, WithScamDB(NewScamDB()), WithRegistry(reg))
|
||||
|
||||
app := &Application{
|
||||
UserID: "u-prov",
|
||||
Email: "[email protected]",
|
||||
FirstName: "John",
|
||||
LastName: "Doe",
|
||||
Phone: "5551234567",
|
||||
DateOfBirth: "1990-01-15",
|
||||
AddressLine1: "123 Main St",
|
||||
City: "New York",
|
||||
State: "NY",
|
||||
ZipCode: "10001",
|
||||
Country: "USA",
|
||||
Status: AppSubmitted,
|
||||
Steps: newApplicationSteps(),
|
||||
}
|
||||
store.SaveApplication(app)
|
||||
|
||||
w := doRequest(router, "POST", "/applications/"+app.ID+"/review", map[string]string{
|
||||
"decision": "approved",
|
||||
})
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
|
||||
var reviewed Application
|
||||
decodeJSON(t, w, &reviewed)
|
||||
if reviewed.Status != AppApproved {
|
||||
t.Fatalf("expected approved, got %s", reviewed.Status)
|
||||
}
|
||||
|
||||
// Wait for async provisioning goroutine (with timeout).
|
||||
select {
|
||||
case req := <-mock.created:
|
||||
if req.Identity.GivenName != "John" {
|
||||
t.Fatalf("expected GivenName John, got %s", req.Identity.GivenName)
|
||||
}
|
||||
if req.Identity.FamilyName != "Doe" {
|
||||
t.Fatalf("expected FamilyName Doe, got %s", req.Identity.FamilyName)
|
||||
}
|
||||
if req.Contact.Email != "[email protected]" {
|
||||
t.Fatalf("expected email [email protected], got %s", req.Contact.Email)
|
||||
}
|
||||
if req.Contact.City != "New York" {
|
||||
t.Fatalf("expected city New York, got %s", req.Contact.City)
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("timed out waiting for Alpaca account provisioning")
|
||||
}
|
||||
|
||||
// Verify the AlpacaAccountID was saved back to the application.
|
||||
// Give the goroutine a moment to complete the save after CreateAccount returns.
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
saved, err := store.GetApplicationByUser("u-prov")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to fetch application: %v", err)
|
||||
}
|
||||
if saved.AlpacaAccountID != "mock-alpaca-acct-001" {
|
||||
t.Fatalf("expected AlpacaAccountID mock-alpaca-acct-001, got %q", saved.AlpacaAccountID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplicationRejectionDoesNotProvision(t *testing.T) {
|
||||
mock := &mockAlpacaProvider{created: make(chan *types.CreateAccountRequest, 1)}
|
||||
reg := provider.NewRegistry()
|
||||
reg.Register(mock)
|
||||
|
||||
store := NewStore()
|
||||
router := NewRouter(store, WithScamDB(NewScamDB()), WithRegistry(reg))
|
||||
|
||||
app := &Application{
|
||||
UserID: "u-rej",
|
||||
Email: "[email protected]",
|
||||
Status: AppSubmitted,
|
||||
Steps: newApplicationSteps(),
|
||||
}
|
||||
store.SaveApplication(app)
|
||||
|
||||
w := doRequest(router, "POST", "/applications/"+app.ID+"/review", map[string]string{
|
||||
"decision": "rejected",
|
||||
})
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("expected 200, got %d", w.Code)
|
||||
}
|
||||
|
||||
// Should NOT trigger provisioning.
|
||||
select {
|
||||
case <-mock.created:
|
||||
t.Fatal("Alpaca account provisioned on rejection - should not happen")
|
||||
case <-time.After(200 * time.Millisecond):
|
||||
// Expected: no provisioning on rejection.
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1192,9 +1192,10 @@ func (s *PostgresStore) SaveApplication(app *Application) error {
|
||||
INSERT INTO applications (id, user_id, email, first_name, last_name, phone, date_of_birth,
|
||||
ssn_hash, ssn_last4, address_line1, address_line2, city, state, zip_code, country,
|
||||
status, current_step, kyc_status, aml_status, steps, documents,
|
||||
risk_level, risk_score, submitted_at, reviewed_by, reviewed_at, created_at, updated_at)
|
||||
risk_level, risk_score, submitted_at, reviewed_by, reviewed_at,
|
||||
alpaca_account_id, created_at, updated_at)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15,
|
||||
$16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28)
|
||||
$16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29)
|
||||
ON CONFLICT (id) DO UPDATE SET
|
||||
email = EXCLUDED.email,
|
||||
first_name = EXCLUDED.first_name,
|
||||
@@ -1220,13 +1221,14 @@ func (s *PostgresStore) SaveApplication(app *Application) error {
|
||||
submitted_at = EXCLUDED.submitted_at,
|
||||
reviewed_by = EXCLUDED.reviewed_by,
|
||||
reviewed_at = EXCLUDED.reviewed_at,
|
||||
alpaca_account_id = EXCLUDED.alpaca_account_id,
|
||||
updated_at = EXCLUDED.updated_at
|
||||
`, app.ID, app.UserID, app.Email, app.FirstName, app.LastName, app.Phone, app.DateOfBirth,
|
||||
app.SSNHash, app.SSNLast4, app.AddressLine1, app.AddressLine2, app.City, app.State,
|
||||
app.ZipCode, app.Country, string(app.Status), app.CurrentStep,
|
||||
string(app.KYCStatus), string(app.AMLStatus), stepsJSON, docsJSON,
|
||||
string(app.RiskLevel), app.RiskScore, submittedAt, app.ReviewedBy, reviewedAt,
|
||||
app.CreatedAt, app.UpdatedAt)
|
||||
app.AlpacaAccountID, app.CreatedAt, app.UpdatedAt)
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -1236,7 +1238,8 @@ func (s *PostgresStore) GetApplication(id string) (*Application, error) {
|
||||
SELECT id, user_id, email, first_name, last_name, phone, date_of_birth,
|
||||
ssn_hash, ssn_last4, address_line1, address_line2, city, state, zip_code, country,
|
||||
status, current_step, kyc_status, aml_status, steps, documents,
|
||||
risk_level, risk_score, submitted_at, reviewed_by, reviewed_at, created_at, updated_at
|
||||
risk_level, risk_score, submitted_at, reviewed_by, reviewed_at,
|
||||
alpaca_account_id, created_at, updated_at
|
||||
FROM applications WHERE id = $1
|
||||
`, id)
|
||||
}
|
||||
@@ -1247,7 +1250,8 @@ func (s *PostgresStore) GetApplicationByUser(userID string) (*Application, error
|
||||
SELECT id, user_id, email, first_name, last_name, phone, date_of_birth,
|
||||
ssn_hash, ssn_last4, address_line1, address_line2, city, state, zip_code, country,
|
||||
status, current_step, kyc_status, aml_status, steps, documents,
|
||||
risk_level, risk_score, submitted_at, reviewed_by, reviewed_at, created_at, updated_at
|
||||
risk_level, risk_score, submitted_at, reviewed_by, reviewed_at,
|
||||
alpaca_account_id, created_at, updated_at
|
||||
FROM applications WHERE user_id = $1 ORDER BY created_at DESC LIMIT 1
|
||||
`, userID)
|
||||
}
|
||||
@@ -1265,7 +1269,7 @@ func (s *PostgresStore) scanApplication(ctx context.Context, query string, args
|
||||
&app.ZipCode, &app.Country, &status, &app.CurrentStep,
|
||||
&kycStatus, &amlStatus, &stepsJSON, &docsJSON,
|
||||
&riskLevel, &app.RiskScore, &submittedAt, &app.ReviewedBy,
|
||||
&reviewedAt, &app.CreatedAt, &app.UpdatedAt,
|
||||
&reviewedAt, &app.AlpacaAccountID, &app.CreatedAt, &app.UpdatedAt,
|
||||
)
|
||||
if err != nil {
|
||||
if err == pgx.ErrNoRows {
|
||||
@@ -1298,7 +1302,8 @@ func (s *PostgresStore) ListApplications() []*Application {
|
||||
SELECT id, user_id, email, first_name, last_name, phone, date_of_birth,
|
||||
ssn_hash, ssn_last4, address_line1, address_line2, city, state, zip_code, country,
|
||||
status, current_step, kyc_status, aml_status, steps, documents,
|
||||
risk_level, risk_score, submitted_at, reviewed_by, reviewed_at, created_at, updated_at
|
||||
risk_level, risk_score, submitted_at, reviewed_by, reviewed_at,
|
||||
alpaca_account_id, created_at, updated_at
|
||||
FROM applications ORDER BY created_at DESC
|
||||
`)
|
||||
if err != nil {
|
||||
@@ -1314,7 +1319,8 @@ func (s *PostgresStore) ListApplicationsByStatus(status ApplicationStatus) []*Ap
|
||||
SELECT id, user_id, email, first_name, last_name, phone, date_of_birth,
|
||||
ssn_hash, ssn_last4, address_line1, address_line2, city, state, zip_code, country,
|
||||
status, current_step, kyc_status, aml_status, steps, documents,
|
||||
risk_level, risk_score, submitted_at, reviewed_by, reviewed_at, created_at, updated_at
|
||||
risk_level, risk_score, submitted_at, reviewed_by, reviewed_at,
|
||||
alpaca_account_id, created_at, updated_at
|
||||
FROM applications WHERE status = $1 ORDER BY created_at DESC
|
||||
`, string(status))
|
||||
if err != nil {
|
||||
@@ -1339,7 +1345,7 @@ func (s *PostgresStore) scanApplicationRows(rows pgx.Rows) []*Application {
|
||||
&app.ZipCode, &app.Country, &status, &app.CurrentStep,
|
||||
&kycStatus, &amlStatus, &stepsJSON, &docsJSON,
|
||||
&riskLevel, &app.RiskScore, &submittedAt, &app.ReviewedBy,
|
||||
&reviewedAt, &app.CreatedAt, &app.UpdatedAt,
|
||||
&reviewedAt, &app.AlpacaAccountID, &app.CreatedAt, &app.UpdatedAt,
|
||||
); err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
"github.com/luxfi/broker/pkg/auth"
|
||||
"github.com/luxfi/broker/pkg/provider"
|
||||
"github.com/luxfi/compliance/pkg/jube"
|
||||
)
|
||||
|
||||
@@ -20,6 +21,7 @@ type RouterOption func(*routerConfig)
|
||||
type routerConfig struct {
|
||||
jubeClient *jube.Client
|
||||
scamDB *ScamDB
|
||||
registry *provider.Registry
|
||||
}
|
||||
|
||||
// WithJubeClient adds the Jube AML sidecar client for live screening.
|
||||
@@ -32,6 +34,11 @@ func WithScamDB(db *ScamDB) RouterOption {
|
||||
return func(cfg *routerConfig) { cfg.scamDB = db }
|
||||
}
|
||||
|
||||
// WithRegistry adds the provider registry for post-approval account provisioning.
|
||||
func WithRegistry(r *provider.Registry) RouterOption {
|
||||
return func(cfg *routerConfig) { cfg.registry = r }
|
||||
}
|
||||
|
||||
// NewRouter creates a chi sub-router with all compliance endpoints.
|
||||
// Mount this under /compliance on the main router.
|
||||
// The adminStore parameter provides authentication and RBAC. When non-nil,
|
||||
@@ -61,7 +68,7 @@ func NewRouter(store ComplianceStore, opts ...RouterOption) chi.Router {
|
||||
creds := &credentialsHandler{store: store}
|
||||
billing := &billingHandler{}
|
||||
aml := &amlHandler{store: store, jubeClient: cfg.jubeClient, scamDB: cfg.scamDB}
|
||||
apps := &applicationHandler{store: store}
|
||||
apps := &applicationHandler{store: store, registry: cfg.registry}
|
||||
|
||||
// Compliance endpoints are restricted to the admin org (built-in).
|
||||
// Customer orgs (liquidity, etc.) access the platform via the exchange app,
|
||||
|
||||
@@ -230,6 +230,10 @@ var migrations = []migration{
|
||||
CREATE INDEX IF NOT EXISTS idx_applications_email ON applications(email);
|
||||
CREATE INDEX IF NOT EXISTS idx_applications_status ON applications(status)`,
|
||||
},
|
||||
{
|
||||
name: "018_add_alpaca_account_id",
|
||||
sql: `ALTER TABLE applications ADD COLUMN IF NOT EXISTS alpaca_account_id TEXT NOT NULL DEFAULT ''`,
|
||||
},
|
||||
}
|
||||
|
||||
// RunMigrations creates the schema_migrations tracking table and runs all
|
||||
|
||||
Reference in New Issue
Block a user