426 lines
17 KiB
Go
426 lines
17 KiB
Go
package customer
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"errors"
|
|
"fmt"
|
|
"net/url"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"time"
|
|
|
|
_ "modernc.org/sqlite"
|
|
)
|
|
|
|
const schema = `
|
|
PRAGMA foreign_keys=ON;
|
|
CREATE TABLE IF NOT EXISTS customers(
|
|
id TEXT PRIMARY KEY,
|
|
username TEXT NOT NULL UNIQUE COLLATE NOCASE,
|
|
password_salt TEXT NOT NULL,
|
|
password_hash TEXT NOT NULL,
|
|
reward_client_id TEXT NOT NULL DEFAULT '',
|
|
created_at INTEGER NOT NULL,
|
|
updated_at INTEGER NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS customer_sessions(
|
|
id TEXT PRIMARY KEY,
|
|
customer_id TEXT NOT NULL REFERENCES customers(id) ON DELETE CASCADE,
|
|
expires_at INTEGER NOT NULL,
|
|
created_at INTEGER NOT NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS customer_sessions_exp_idx ON customer_sessions(expires_at);
|
|
CREATE TABLE IF NOT EXISTS credit_ledger(
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
customer_id TEXT NOT NULL REFERENCES customers(id) ON DELETE CASCADE,
|
|
delta_micros INTEGER NOT NULL,
|
|
reason TEXT NOT NULL,
|
|
reference TEXT NOT NULL UNIQUE,
|
|
created_at INTEGER NOT NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS credit_ledger_customer_idx ON credit_ledger(customer_id,created_at DESC);
|
|
CREATE TABLE IF NOT EXISTS workers(
|
|
id TEXT PRIMARY KEY,
|
|
customer_id TEXT NOT NULL REFERENCES customers(id) ON DELETE CASCADE,
|
|
task_id TEXT NOT NULL,
|
|
beacon_path TEXT NOT NULL DEFAULT 'auto',
|
|
docker_container_id TEXT NOT NULL DEFAULT '',
|
|
docker_volume TEXT NOT NULL,
|
|
status TEXT NOT NULL DEFAULT 'stopped' CHECK(status IN ('stopped','starting','running','error')),
|
|
worker_client_id TEXT NOT NULL DEFAULT '',
|
|
register_token TEXT NOT NULL,
|
|
rate_micros_per_minute INTEGER NOT NULL,
|
|
last_charge_at INTEGER,
|
|
created_at INTEGER NOT NULL,
|
|
updated_at INTEGER NOT NULL,
|
|
last_error TEXT NOT NULL DEFAULT ''
|
|
);
|
|
CREATE INDEX IF NOT EXISTS workers_customer_idx ON workers(customer_id,created_at);
|
|
CREATE TABLE IF NOT EXISTS paypal_orders(
|
|
order_id TEXT PRIMARY KEY,
|
|
customer_id TEXT NOT NULL REFERENCES customers(id) ON DELETE CASCADE,
|
|
package_id TEXT NOT NULL,
|
|
amount_cents INTEGER NOT NULL,
|
|
currency TEXT NOT NULL,
|
|
credits_micros INTEGER NOT NULL,
|
|
status TEXT NOT NULL,
|
|
capture_id TEXT NOT NULL DEFAULT '',
|
|
created_at INTEGER NOT NULL,
|
|
updated_at INTEGER NOT NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS paypal_orders_customer_idx ON paypal_orders(customer_id,created_at DESC);
|
|
`
|
|
|
|
type Store struct{ DB *sql.DB }
|
|
|
|
func Open(ctx context.Context, path string) (*Store, error) {
|
|
if strings.TrimSpace(path) == "" {
|
|
path = "/customer-data/customer-service.db"
|
|
}
|
|
abs, err := filepath.Abs(path)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if err := os.MkdirAll(filepath.Dir(abs), 0o750); err != nil {
|
|
return nil, err
|
|
}
|
|
u := &url.URL{Scheme: "file", Path: filepath.ToSlash(abs)}
|
|
q := u.Query()
|
|
q.Add("_pragma", "busy_timeout(10000)")
|
|
q.Add("_pragma", "foreign_keys(ON)")
|
|
q.Add("_pragma", "synchronous(NORMAL)")
|
|
q.Set("_txlock", "immediate")
|
|
u.RawQuery = q.Encode()
|
|
db, err := sql.Open("sqlite", u.String())
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
db.SetMaxOpenConns(4)
|
|
if err := db.PingContext(ctx); err != nil {
|
|
db.Close()
|
|
return nil, err
|
|
}
|
|
if _, err := db.ExecContext(ctx, "PRAGMA journal_mode=WAL"); err != nil {
|
|
db.Close()
|
|
return nil, err
|
|
}
|
|
for i, stmt := range strings.Split(schema, ";") {
|
|
stmt = strings.TrimSpace(stmt)
|
|
if stmt == "" {
|
|
continue
|
|
}
|
|
if _, err := db.ExecContext(ctx, stmt); err != nil {
|
|
db.Close()
|
|
return nil, fmt.Errorf("customer schema %d: %w", i+1, err)
|
|
}
|
|
}
|
|
return &Store{DB: db}, nil
|
|
}
|
|
|
|
type Customer struct {
|
|
ID string `json:"id"`
|
|
Username string `json:"username"`
|
|
RewardClientID string `json:"reward_client_id"`
|
|
CreatedAt time.Time `json:"created_at"`
|
|
}
|
|
|
|
type Worker struct {
|
|
ID string `json:"id"`
|
|
CustomerID string `json:"customer_id"`
|
|
TaskID string `json:"task_id"`
|
|
BeaconPath string `json:"beacon_path"`
|
|
ContainerID string `json:"container_id"`
|
|
Volume string `json:"volume"`
|
|
Status string `json:"status"`
|
|
WorkerClientID string `json:"worker_client_id"`
|
|
RegisterToken string `json:"-"`
|
|
RateMicrosPerMinute int64 `json:"rate_micros_per_minute"`
|
|
LastChargeAt *time.Time `json:"last_charge_at,omitempty"`
|
|
CreatedAt time.Time `json:"created_at"`
|
|
UpdatedAt time.Time `json:"updated_at"`
|
|
LastError string `json:"last_error,omitempty"`
|
|
}
|
|
|
|
func (s *Store) CreateCustomer(ctx context.Context, id, username, salt, hash string) error {
|
|
now := time.Now().UTC().UnixMilli()
|
|
_, err := s.DB.ExecContext(ctx, `INSERT INTO customers(id,username,password_salt,password_hash,created_at,updated_at) VALUES(?,?,?,?,?,?)`, id, strings.TrimSpace(username), salt, hash, now, now)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) CustomerByUsername(ctx context.Context, username string) (Customer, string, string, error) {
|
|
var c Customer
|
|
var salt, hash string
|
|
var created int64
|
|
err := s.DB.QueryRowContext(ctx, `SELECT id,username,reward_client_id,created_at,password_salt,password_hash FROM customers WHERE username=?`, strings.TrimSpace(username)).Scan(&c.ID, &c.Username, &c.RewardClientID, &created, &salt, &hash)
|
|
c.CreatedAt = time.UnixMilli(created).UTC()
|
|
return c, salt, hash, err
|
|
}
|
|
func (s *Store) CustomerByID(ctx context.Context, id string) (Customer, error) {
|
|
var c Customer
|
|
var created int64
|
|
err := s.DB.QueryRowContext(ctx, `SELECT id,username,reward_client_id,created_at FROM customers WHERE id=?`, id).Scan(&c.ID, &c.Username, &c.RewardClientID, &created)
|
|
c.CreatedAt = time.UnixMilli(created).UTC()
|
|
return c, err
|
|
}
|
|
func (s *Store) SetRewardClientID(ctx context.Context, id, cid string) error {
|
|
_, err := s.DB.ExecContext(ctx, `UPDATE customers SET reward_client_id=?,updated_at=? WHERE id=?`, strings.TrimSpace(cid), time.Now().UTC().UnixMilli(), id)
|
|
return err
|
|
}
|
|
|
|
func (s *Store) CreateSession(ctx context.Context, sid, cid string, ttl time.Duration) error {
|
|
now := time.Now().UTC()
|
|
_, err := s.DB.ExecContext(ctx, `INSERT INTO customer_sessions(id,customer_id,expires_at,created_at) VALUES(?,?,?,?)`, sid, cid, now.Add(ttl).UnixMilli(), now.UnixMilli())
|
|
return err
|
|
}
|
|
func (s *Store) SessionCustomer(ctx context.Context, sid string) (string, error) {
|
|
var cid string
|
|
err := s.DB.QueryRowContext(ctx, `SELECT customer_id FROM customer_sessions WHERE id=? AND expires_at>?`, sid, time.Now().UTC().UnixMilli()).Scan(&cid)
|
|
return cid, err
|
|
}
|
|
func (s *Store) DeleteSession(ctx context.Context, sid string) {
|
|
_, _ = s.DB.ExecContext(ctx, `DELETE FROM customer_sessions WHERE id=?`, sid)
|
|
}
|
|
|
|
func (s *Store) BalanceMicros(ctx context.Context, cid string) (int64, error) {
|
|
var v sql.NullInt64
|
|
err := s.DB.QueryRowContext(ctx, `SELECT sum(delta_micros) FROM credit_ledger WHERE customer_id=?`, cid).Scan(&v)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return v.Int64, nil
|
|
}
|
|
func (s *Store) AddLedger(ctx context.Context, cid string, delta int64, reason, ref string) error {
|
|
if delta == 0 {
|
|
return errors.New("zero credit change")
|
|
}
|
|
_, err := s.DB.ExecContext(ctx, `INSERT INTO credit_ledger(customer_id,delta_micros,reason,reference,created_at) VALUES(?,?,?,?,?)`, cid, delta, reason, ref, time.Now().UTC().UnixMilli())
|
|
return err
|
|
}
|
|
|
|
type LedgerItem struct {
|
|
DeltaMicros int64 `json:"delta_micros"`
|
|
Reason string `json:"reason"`
|
|
Reference string `json:"reference"`
|
|
CreatedAt time.Time `json:"created_at"`
|
|
}
|
|
|
|
func (s *Store) Ledger(ctx context.Context, cid string, limit int) ([]LedgerItem, error) {
|
|
if limit < 1 || limit > 200 {
|
|
limit = 50
|
|
}
|
|
rows, err := s.DB.QueryContext(ctx, `SELECT delta_micros,reason,reference,created_at FROM credit_ledger WHERE customer_id=? ORDER BY id DESC LIMIT ?`, cid, limit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var out []LedgerItem
|
|
for rows.Next() {
|
|
var x LedgerItem
|
|
var ms int64
|
|
if err := rows.Scan(&x.DeltaMicros, &x.Reason, &x.Reference, &ms); err != nil {
|
|
return nil, err
|
|
}
|
|
x.CreatedAt = time.UnixMilli(ms).UTC()
|
|
out = append(out, x)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
func scanWorker(row interface{ Scan(...any) error }) (Worker, error) {
|
|
var w Worker
|
|
var last sql.NullInt64
|
|
var created, updated int64
|
|
err := row.Scan(&w.ID, &w.CustomerID, &w.TaskID, &w.BeaconPath, &w.ContainerID, &w.Volume, &w.Status, &w.WorkerClientID, &w.RegisterToken, &w.RateMicrosPerMinute, &last, &created, &updated, &w.LastError)
|
|
if err != nil {
|
|
return w, err
|
|
}
|
|
if last.Valid {
|
|
v := time.UnixMilli(last.Int64).UTC()
|
|
w.LastChargeAt = &v
|
|
}
|
|
w.CreatedAt = time.UnixMilli(created).UTC()
|
|
w.UpdatedAt = time.UnixMilli(updated).UTC()
|
|
return w, nil
|
|
}
|
|
|
|
const workerCols = `id,customer_id,task_id,beacon_path,docker_container_id,docker_volume,status,worker_client_id,register_token,rate_micros_per_minute,last_charge_at,created_at,updated_at,last_error`
|
|
|
|
func (s *Store) CreateWorker(ctx context.Context, w Worker) error {
|
|
now := time.Now().UTC().UnixMilli()
|
|
_, err := s.DB.ExecContext(ctx, `INSERT INTO workers(id,customer_id,task_id,beacon_path,docker_volume,status,register_token,rate_micros_per_minute,created_at,updated_at) VALUES(?,?,?,?,?,'stopped',?,?,?,?)`, w.ID, w.CustomerID, w.TaskID, w.BeaconPath, w.Volume, w.RegisterToken, w.RateMicrosPerMinute, now, now)
|
|
return err
|
|
}
|
|
func (s *Store) Worker(ctx context.Context, cid, wid string) (Worker, error) {
|
|
return scanWorker(s.DB.QueryRowContext(ctx, `SELECT `+workerCols+` FROM workers WHERE id=? AND customer_id=?`, wid, cid))
|
|
}
|
|
func (s *Store) WorkerByID(ctx context.Context, wid string) (Worker, error) {
|
|
return scanWorker(s.DB.QueryRowContext(ctx, `SELECT `+workerCols+` FROM workers WHERE id=?`, wid))
|
|
}
|
|
func (s *Store) Workers(ctx context.Context, cid string) ([]Worker, error) {
|
|
rows, err := s.DB.QueryContext(ctx, `SELECT `+workerCols+` FROM workers WHERE customer_id=? ORDER BY created_at`, cid)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var out []Worker
|
|
for rows.Next() {
|
|
w, err := scanWorker(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, w)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
func (s *Store) WorkerCount(ctx context.Context, cid string, runningOnly bool) (int, error) {
|
|
q := `SELECT count(*) FROM workers WHERE customer_id=?`
|
|
if runningOnly {
|
|
q += ` AND status='running'`
|
|
}
|
|
var n int
|
|
err := s.DB.QueryRowContext(ctx, q, cid).Scan(&n)
|
|
return n, err
|
|
}
|
|
func (s *Store) TotalWorkerCount(ctx context.Context) (int, error) {
|
|
var n int
|
|
err := s.DB.QueryRowContext(ctx, `SELECT count(*) FROM workers`).Scan(&n)
|
|
return n, err
|
|
}
|
|
|
|
func (s *Store) RunningWorkerCount(ctx context.Context) (int, error) {
|
|
var n int
|
|
err := s.DB.QueryRowContext(ctx, `SELECT count(*) FROM workers WHERE status='running'`).Scan(&n)
|
|
return n, err
|
|
}
|
|
|
|
func (s *Store) RunningWorkers(ctx context.Context) ([]Worker, error) {
|
|
rows, err := s.DB.QueryContext(ctx, `SELECT `+workerCols+` FROM workers WHERE status='running' ORDER BY created_at`)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var out []Worker
|
|
for rows.Next() {
|
|
w, err := scanWorker(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, w)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
func (s *Store) ClaimWorkerStart(ctx context.Context, cid, wid string) (bool, error) {
|
|
res, err := s.DB.ExecContext(ctx, `UPDATE workers SET status='starting',last_error='',updated_at=? WHERE id=? AND customer_id=? AND status IN ('stopped','error')`, time.Now().UTC().UnixMilli(), wid, cid)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
n, err := res.RowsAffected()
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
return n == 1, nil
|
|
}
|
|
func (s *Store) RecoverStartingWorkers(ctx context.Context) error {
|
|
_, err := s.DB.ExecContext(ctx, `UPDATE workers SET status='stopped',last_error='recovered after Customer Service restart',updated_at=? WHERE status='starting'`, time.Now().UTC().UnixMilli())
|
|
return err
|
|
}
|
|
|
|
func (s *Store) SetWorkerRuntime(ctx context.Context, wid, status, containerID, lastErr string) error {
|
|
_, err := s.DB.ExecContext(ctx, `UPDATE workers SET status=?,docker_container_id=?,last_error=?,updated_at=? WHERE id=?`, status, containerID, lastErr, time.Now().UTC().UnixMilli(), wid)
|
|
return err
|
|
}
|
|
func (s *Store) SetWorkerClient(ctx context.Context, wid, clientID string) error {
|
|
_, err := s.DB.ExecContext(ctx, `UPDATE workers SET worker_client_id=?,updated_at=? WHERE id=?`, clientID, time.Now().UTC().UnixMilli(), wid)
|
|
return err
|
|
}
|
|
func (s *Store) UpdateWorkerConfig(ctx context.Context, cid, wid, taskID, beaconPath string) error {
|
|
_, err := s.DB.ExecContext(ctx, `UPDATE workers SET task_id=?,beacon_path=?,updated_at=? WHERE id=? AND customer_id=?`, taskID, beaconPath, time.Now().UTC().UnixMilli(), wid, cid)
|
|
return err
|
|
}
|
|
func (s *Store) DeleteWorker(ctx context.Context, cid, wid string) error {
|
|
_, err := s.DB.ExecContext(ctx, `DELETE FROM workers WHERE id=? AND customer_id=?`, wid, cid)
|
|
return err
|
|
}
|
|
func (s *Store) MarkWorkerCharged(ctx context.Context, wid string, when time.Time) error {
|
|
_, err := s.DB.ExecContext(ctx, `UPDATE workers SET last_charge_at=?,updated_at=? WHERE id=?`, when.UTC().UnixMilli(), time.Now().UTC().UnixMilli(), wid)
|
|
return err
|
|
}
|
|
|
|
// ChargeWorkerMinute debits one prepaid minute atomically. It never allows a
|
|
// negative balance, so a billing loop can stop the worker as soon as funding is
|
|
// exhausted.
|
|
func (s *Store) ChargeWorkerMinute(ctx context.Context, w Worker, minute time.Time) (bool, error) {
|
|
tx, err := s.DB.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
defer tx.Rollback()
|
|
var bal sql.NullInt64
|
|
if err := tx.QueryRowContext(ctx, `SELECT sum(delta_micros) FROM credit_ledger WHERE customer_id=?`, w.CustomerID).Scan(&bal); err != nil {
|
|
return false, err
|
|
}
|
|
if bal.Int64 < w.RateMicrosPerMinute {
|
|
return false, nil
|
|
}
|
|
ref := fmt.Sprintf("worker:%s:%d", w.ID, minute.UTC().UnixMilli())
|
|
if _, err := tx.ExecContext(ctx, `INSERT OR IGNORE INTO credit_ledger(customer_id,delta_micros,reason,reference,created_at) VALUES(?,?,?,?,?)`, w.CustomerID, -w.RateMicrosPerMinute, "worker_minute", ref, time.Now().UTC().UnixMilli()); err != nil {
|
|
return false, err
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `UPDATE workers SET last_charge_at=?,updated_at=? WHERE id=?`, minute.UTC().UnixMilli(), time.Now().UTC().UnixMilli(), w.ID); err != nil {
|
|
return false, err
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return false, err
|
|
}
|
|
return true, nil
|
|
}
|
|
|
|
func (s *Store) RefundWorkerStartMinute(ctx context.Context, w Worker, chargedAt time.Time, detail string) error {
|
|
ref := fmt.Sprintf("worker_start_refund:%s:%d", w.ID, chargedAt.UTC().UnixMilli())
|
|
reason := "worker_start_refund"
|
|
if strings.TrimSpace(detail) != "" {
|
|
reason += ":" + strings.TrimSpace(detail)
|
|
}
|
|
_, err := s.DB.ExecContext(ctx, `INSERT OR IGNORE INTO credit_ledger(customer_id,delta_micros,reason,reference,created_at) VALUES(?,?,?,?,?)`, w.CustomerID, w.RateMicrosPerMinute, reason, ref, time.Now().UTC().UnixMilli())
|
|
return err
|
|
}
|
|
|
|
func (s *Store) UpsertPayPalOrder(ctx context.Context, orderID, cid, pkg string, cents int64, currency string, credits int64, status string) error {
|
|
now := time.Now().UTC().UnixMilli()
|
|
_, err := s.DB.ExecContext(ctx, `INSERT INTO paypal_orders(order_id,customer_id,package_id,amount_cents,currency,credits_micros,status,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?) ON CONFLICT(order_id) DO UPDATE SET status=excluded.status,updated_at=excluded.updated_at`, orderID, cid, pkg, cents, currency, credits, status, now, now)
|
|
return err
|
|
}
|
|
|
|
type PayPalOrder struct {
|
|
OrderID, CustomerID, PackageID, Currency, Status, CaptureID string
|
|
AmountCents, CreditsMicros int64
|
|
}
|
|
|
|
func (s *Store) PayPalOrder(ctx context.Context, orderID string) (PayPalOrder, error) {
|
|
var o PayPalOrder
|
|
err := s.DB.QueryRowContext(ctx, `SELECT order_id,customer_id,package_id,amount_cents,currency,credits_micros,status,capture_id FROM paypal_orders WHERE order_id=?`, orderID).Scan(&o.OrderID, &o.CustomerID, &o.PackageID, &o.AmountCents, &o.Currency, &o.CreditsMicros, &o.Status, &o.CaptureID)
|
|
return o, err
|
|
}
|
|
func (s *Store) CompletePayPalOrder(ctx context.Context, orderID, captureID string) error {
|
|
tx, err := s.DB.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
var o PayPalOrder
|
|
if err := tx.QueryRowContext(ctx, `SELECT order_id,customer_id,package_id,amount_cents,currency,credits_micros,status,capture_id FROM paypal_orders WHERE order_id=?`, orderID).Scan(&o.OrderID, &o.CustomerID, &o.PackageID, &o.AmountCents, &o.Currency, &o.CreditsMicros, &o.Status, &o.CaptureID); err != nil {
|
|
return err
|
|
}
|
|
ref := "paypal:" + orderID
|
|
if _, err := tx.ExecContext(ctx, `INSERT OR IGNORE INTO credit_ledger(customer_id,delta_micros,reason,reference,created_at) VALUES(?,?,?,?,?)`, o.CustomerID, o.CreditsMicros, "paypal_topup", ref, time.Now().UTC().UnixMilli()); err != nil {
|
|
return err
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `UPDATE paypal_orders SET status='COMPLETED',capture_id=?,updated_at=? WHERE order_id=?`, captureID, time.Now().UTC().UnixMilli(), orderID); err != nil {
|
|
return err
|
|
}
|
|
return tx.Commit()
|
|
}
|