761 lines
31 KiB
Go
761 lines
31 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 '',
|
|
blocked INTEGER NOT NULL DEFAULT 0,
|
|
blocked_reason TEXT NOT NULL DEFAULT '',
|
|
blocked_at INTEGER,
|
|
worker_limit_bypass INTEGER NOT NULL DEFAULT 0,
|
|
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 '',
|
|
controller_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,
|
|
last_lease_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 service_controllers(
|
|
id TEXT PRIMARY KEY,
|
|
name TEXT NOT NULL,
|
|
base_url TEXT NOT NULL,
|
|
protocol_version INTEGER NOT NULL DEFAULT 1,
|
|
max_workers INTEGER NOT NULL DEFAULT 1000,
|
|
max_running INTEGER NOT NULL DEFAULT 100,
|
|
reported_workers INTEGER NOT NULL DEFAULT 0,
|
|
reported_running INTEGER NOT NULL DEFAULT 0,
|
|
enabled INTEGER NOT NULL DEFAULT 1,
|
|
draining INTEGER NOT NULL DEFAULT 0,
|
|
last_seen INTEGER NOT NULL,
|
|
last_error TEXT NOT NULL DEFAULT '',
|
|
created_at INTEGER NOT NULL,
|
|
updated_at INTEGER NOT NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS service_controllers_seen_idx ON service_controllers(last_seen);
|
|
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);
|
|
CREATE TABLE IF NOT EXISTS customer_settings(
|
|
key TEXT PRIMARY KEY,
|
|
value TEXT NOT NULL,
|
|
updated_at INTEGER NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS registration_invites(
|
|
token_hash TEXT PRIMARY KEY,
|
|
label TEXT NOT NULL DEFAULT '',
|
|
expires_at INTEGER NOT NULL,
|
|
used_by TEXT REFERENCES customers(id) ON DELETE SET NULL,
|
|
used_at INTEGER,
|
|
created_at INTEGER NOT NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS registration_invites_exp_idx ON registration_invites(expires_at,used_at);
|
|
`
|
|
|
|
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)
|
|
}
|
|
}
|
|
// Existing V4 customer databases need explicit column evolution because
|
|
// CREATE TABLE IF NOT EXISTS does not alter an already-created table.
|
|
for _, m := range []struct{ name, def string }{
|
|
{"blocked", "INTEGER NOT NULL DEFAULT 0"},
|
|
{"blocked_reason", "TEXT NOT NULL DEFAULT ''"},
|
|
{"blocked_at", "INTEGER"},
|
|
{"worker_limit_bypass", "INTEGER NOT NULL DEFAULT 0"},
|
|
} {
|
|
if err := ensureCustomerColumn(ctx, db, "customers", m.name, m.def); err != nil {
|
|
db.Close()
|
|
return nil, fmt.Errorf("customer add customers.%s: %w", m.name, err)
|
|
}
|
|
}
|
|
for _, m := range []struct{ name, def string }{
|
|
{"controller_id", "TEXT NOT NULL DEFAULT ''"},
|
|
{"last_lease_at", "INTEGER"},
|
|
} {
|
|
if err := ensureCustomerColumn(ctx, db, "workers", m.name, m.def); err != nil {
|
|
db.Close()
|
|
return nil, fmt.Errorf("customer add workers.%s: %w", m.name, err)
|
|
}
|
|
}
|
|
if err := ensureCustomerColumn(ctx, db, "service_controllers", "protocol_version", "INTEGER NOT NULL DEFAULT 1"); err != nil {
|
|
db.Close()
|
|
return nil, fmt.Errorf("customer add service_controllers.protocol_version: %w", err)
|
|
}
|
|
return &Store{DB: db}, nil
|
|
}
|
|
|
|
func ensureCustomerColumn(ctx context.Context, db *sql.DB, table, name, def string) error {
|
|
rows, err := db.QueryContext(ctx, `PRAGMA table_info(`+table+`)`)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var cid int
|
|
var col, typ string
|
|
var notnull, pk int
|
|
var dflt sql.NullString
|
|
if err := rows.Scan(&cid, &col, &typ, ¬null, &dflt, &pk); err != nil {
|
|
return err
|
|
}
|
|
if strings.EqualFold(col, name) {
|
|
return nil
|
|
}
|
|
}
|
|
_, err = db.ExecContext(ctx, `ALTER TABLE `+table+` ADD COLUMN `+name+` `+def)
|
|
return err
|
|
}
|
|
|
|
type Customer struct {
|
|
ID string `json:"id"`
|
|
Username string `json:"username"`
|
|
RewardClientID string `json:"reward_client_id"`
|
|
Blocked bool `json:"blocked"`
|
|
BlockedReason string `json:"blocked_reason,omitempty"`
|
|
BlockedAt *time.Time `json:"blocked_at,omitempty"`
|
|
WorkerLimitBypass bool `json:"worker_limit_bypass"`
|
|
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"`
|
|
ControllerID string `json:"controller_id,omitempty"`
|
|
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"`
|
|
LastLeaseAt *time.Time `json:"last_lease_at,omitempty"`
|
|
CreatedAt time.Time `json:"created_at"`
|
|
UpdatedAt time.Time `json:"updated_at"`
|
|
LastError string `json:"last_error,omitempty"`
|
|
}
|
|
|
|
type ServiceController struct {
|
|
ID string `json:"id"`
|
|
Name string `json:"name"`
|
|
BaseURL string `json:"base_url"`
|
|
ProtocolVersion int `json:"protocol_version"`
|
|
MaxWorkers int `json:"max_workers"`
|
|
MaxRunning int `json:"max_running"`
|
|
ReportedWorkers int `json:"reported_workers"`
|
|
ReportedRunning int `json:"reported_running"`
|
|
Enabled bool `json:"enabled"`
|
|
Draining bool `json:"draining"`
|
|
LastSeen time.Time `json:"last_seen"`
|
|
LastError string `json:"last_error,omitempty"`
|
|
CreatedAt time.Time `json:"created_at"`
|
|
UpdatedAt time.Time `json:"updated_at"`
|
|
}
|
|
|
|
func scanServiceController(row interface{ Scan(...any) error }) (ServiceController, error) {
|
|
var c ServiceController
|
|
var enabled, draining int
|
|
var seen, created, updated int64
|
|
err := row.Scan(&c.ID, &c.Name, &c.BaseURL, &c.ProtocolVersion, &c.MaxWorkers, &c.MaxRunning, &c.ReportedWorkers, &c.ReportedRunning, &enabled, &draining, &seen, &c.LastError, &created, &updated)
|
|
c.Enabled = enabled != 0
|
|
c.Draining = draining != 0
|
|
c.LastSeen = time.UnixMilli(seen).UTC()
|
|
c.CreatedAt = time.UnixMilli(created).UTC()
|
|
c.UpdatedAt = time.UnixMilli(updated).UTC()
|
|
return c, err
|
|
}
|
|
|
|
const controllerCols = `id,name,base_url,protocol_version,max_workers,max_running,reported_workers,reported_running,enabled,draining,last_seen,last_error,created_at,updated_at`
|
|
|
|
func (s *Store) UpsertServiceController(ctx context.Context, c ServiceController) error {
|
|
now := time.Now().UTC().UnixMilli()
|
|
if c.MaxWorkers <= 0 {
|
|
c.MaxWorkers = 1000
|
|
}
|
|
if c.MaxRunning <= 0 {
|
|
c.MaxRunning = 100
|
|
}
|
|
if c.ProtocolVersion <= 0 {
|
|
c.ProtocolVersion = 1
|
|
}
|
|
_, err := s.DB.ExecContext(ctx, `INSERT INTO service_controllers(id,name,base_url,protocol_version,max_workers,max_running,reported_workers,reported_running,enabled,draining,last_seen,last_error,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,1,0,?,'',?,?)
|
|
ON CONFLICT(id) DO UPDATE SET name=excluded.name,base_url=excluded.base_url,protocol_version=excluded.protocol_version,max_workers=excluded.max_workers,max_running=excluded.max_running,reported_workers=excluded.reported_workers,reported_running=excluded.reported_running,last_seen=excluded.last_seen,last_error='',updated_at=excluded.updated_at`, strings.TrimSpace(c.ID), strings.TrimSpace(c.Name), strings.TrimRight(strings.TrimSpace(c.BaseURL), "/"), c.ProtocolVersion, c.MaxWorkers, c.MaxRunning, c.ReportedWorkers, c.ReportedRunning, now, now, now)
|
|
return err
|
|
}
|
|
func (s *Store) ServiceController(ctx context.Context, id string) (ServiceController, error) {
|
|
return scanServiceController(s.DB.QueryRowContext(ctx, `SELECT `+controllerCols+` FROM service_controllers WHERE id=?`, strings.TrimSpace(id)))
|
|
}
|
|
func (s *Store) ServiceControllers(ctx context.Context) ([]ServiceController, error) {
|
|
rows, err := s.DB.QueryContext(ctx, `SELECT `+controllerCols+` FROM service_controllers ORDER BY name,id`)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var out []ServiceController
|
|
for rows.Next() {
|
|
c, err := scanServiceController(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, c)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
func (s *Store) SetServiceControllerPolicy(ctx context.Context, id string, enabled, draining bool) error {
|
|
_, err := s.DB.ExecContext(ctx, `UPDATE service_controllers SET enabled=?,draining=?,updated_at=? WHERE id=?`, boolInt(enabled), boolInt(draining), time.Now().UTC().UnixMilli(), strings.TrimSpace(id))
|
|
return err
|
|
}
|
|
func (s *Store) ControllerWorkerCounts(ctx context.Context, id string) (total, running int, err error) {
|
|
err = s.DB.QueryRowContext(ctx, `SELECT count(*),COALESCE(sum(CASE WHEN status IN ('running','starting') THEN 1 ELSE 0 END),0) FROM workers WHERE controller_id=?`, strings.TrimSpace(id)).Scan(&total, &running)
|
|
return
|
|
}
|
|
func boolInt(v bool) int {
|
|
if v {
|
|
return 1
|
|
}
|
|
return 0
|
|
}
|
|
|
|
var ErrInvalidInvite = errors.New("registration invite invalid, expired or already used")
|
|
|
|
func scanCustomer(row interface{ Scan(...any) error }, withPassword bool) (Customer, string, string, error) {
|
|
var c Customer
|
|
var blocked, workerLimitBypass int
|
|
var blockedAt sql.NullInt64
|
|
var created int64
|
|
var salt, hash string
|
|
args := []any{&c.ID, &c.Username, &c.RewardClientID, &blocked, &c.BlockedReason, &blockedAt, &workerLimitBypass, &created}
|
|
if withPassword {
|
|
args = append(args, &salt, &hash)
|
|
}
|
|
err := row.Scan(args...)
|
|
c.Blocked = blocked != 0
|
|
c.WorkerLimitBypass = workerLimitBypass != 0
|
|
c.CreatedAt = time.UnixMilli(created).UTC()
|
|
if blockedAt.Valid {
|
|
v := time.UnixMilli(blockedAt.Int64).UTC()
|
|
c.BlockedAt = &v
|
|
}
|
|
return c, salt, hash, err
|
|
}
|
|
|
|
// CreateCustomer atomically creates the account, consumes an optional one-shot
|
|
// invite and grants the configured signup credits. A failed invite therefore
|
|
// cannot leave behind a partially-created account or free ledger entry.
|
|
func (s *Store) CreateCustomer(ctx context.Context, id, username, salt, hash string, signupBonusMicros int64, inviteHash string) error {
|
|
tx, err := s.DB.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
now := time.Now().UTC().UnixMilli()
|
|
if _, err := tx.ExecContext(ctx, `INSERT INTO customers(id,username,password_salt,password_hash,created_at,updated_at) VALUES(?,?,?,?,?,?)`, id, strings.TrimSpace(username), salt, hash, now, now); err != nil {
|
|
return err
|
|
}
|
|
if strings.TrimSpace(inviteHash) != "" {
|
|
res, err := tx.ExecContext(ctx, `UPDATE registration_invites SET used_by=?,used_at=? WHERE token_hash=? AND used_by IS NULL AND expires_at>?`, id, now, inviteHash, now)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
n, _ := res.RowsAffected()
|
|
if n != 1 {
|
|
return ErrInvalidInvite
|
|
}
|
|
}
|
|
if signupBonusMicros > 0 {
|
|
if _, err := tx.ExecContext(ctx, `INSERT INTO credit_ledger(customer_id,delta_micros,reason,reference,created_at) VALUES(?,?,?,?,?)`, id, signupBonusMicros, "signup_bonus", "signup:"+id, now); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return tx.Commit()
|
|
}
|
|
|
|
func (s *Store) CustomerByUsername(ctx context.Context, username string) (Customer, string, string, error) {
|
|
return scanCustomer(s.DB.QueryRowContext(ctx, `SELECT id,username,reward_client_id,blocked,blocked_reason,blocked_at,worker_limit_bypass,created_at,password_salt,password_hash FROM customers WHERE username=?`, strings.TrimSpace(username)), true)
|
|
}
|
|
func (s *Store) CustomerByID(ctx context.Context, id string) (Customer, error) {
|
|
c, _, _, err := scanCustomer(s.DB.QueryRowContext(ctx, `SELECT id,username,reward_client_id,blocked,blocked_reason,blocked_at,worker_limit_bypass,created_at FROM customers WHERE id=?`, id), false)
|
|
return c, err
|
|
}
|
|
func (s *Store) CustomerByRewardClientID(ctx context.Context, rewardClientID string) (Customer, error) {
|
|
c, _, _, err := scanCustomer(s.DB.QueryRowContext(ctx, `SELECT id,username,reward_client_id,blocked,blocked_reason,blocked_at,worker_limit_bypass,created_at FROM customers WHERE reward_client_id=? ORDER BY created_at LIMIT 1`, strings.TrimSpace(rewardClientID)), false)
|
|
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) SetCustomerBlocked(ctx context.Context, id string, blocked bool, reason string) error {
|
|
now := time.Now().UTC().UnixMilli()
|
|
if blocked {
|
|
_, err := s.DB.ExecContext(ctx, `UPDATE customers SET blocked=1,blocked_reason=?,blocked_at=?,updated_at=? WHERE id=?`, strings.TrimSpace(reason), now, now, id)
|
|
return err
|
|
}
|
|
_, err := s.DB.ExecContext(ctx, `UPDATE customers SET blocked=0,blocked_reason='',blocked_at=NULL,updated_at=? WHERE id=?`, now, id)
|
|
return err
|
|
}
|
|
func (s *Store) SetCustomerWorkerLimitBypass(ctx context.Context, id string, bypass bool) error {
|
|
v := 0
|
|
if bypass {
|
|
v = 1
|
|
}
|
|
res, err := s.DB.ExecContext(ctx, `UPDATE customers SET worker_limit_bypass=?,updated_at=? WHERE id=?`, v, time.Now().UTC().UnixMilli(), id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
n, _ := res.RowsAffected()
|
|
if n == 0 {
|
|
return sql.ErrNoRows
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) DeleteCustomerSessions(ctx context.Context, cid string) error {
|
|
_, err := s.DB.ExecContext(ctx, `DELETE FROM customer_sessions WHERE customer_id=?`, cid)
|
|
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 s.customer_id FROM customer_sessions s JOIN customers c ON c.id=s.customer_id WHERE s.id=? AND s.expires_at>? AND c.blocked=0`, 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()
|
|
}
|
|
|
|
type LedgerSummaryItem struct {
|
|
DeltaMicros int64 `json:"delta_micros"`
|
|
Reason string `json:"reason"`
|
|
Count int `json:"count"`
|
|
CreatedAt time.Time `json:"created_at"`
|
|
Day string `json:"day"`
|
|
}
|
|
|
|
// LedgerSummary intentionally collapses high-frequency worker-minute and
|
|
// positive-tip entries into daily reason buckets. The raw ledger stays intact
|
|
// in SQLite for auditing and idempotency.
|
|
func (s *Store) LedgerSummary(ctx context.Context, cid string, limit int) ([]LedgerSummaryItem, error) {
|
|
if limit < 1 || limit > 120 {
|
|
limit = 45
|
|
}
|
|
rows, err := s.DB.QueryContext(ctx, `SELECT reason,COALESCE(sum(delta_micros),0),count(*),max(created_at),strftime('%Y-%m-%d',max(created_at)/1000,'unixepoch')
|
|
FROM credit_ledger WHERE customer_id=?
|
|
GROUP BY reason,strftime('%Y-%m-%d',created_at/1000,'unixepoch')
|
|
ORDER BY max(created_at) DESC LIMIT ?`, cid, limit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var out []LedgerSummaryItem
|
|
for rows.Next() {
|
|
var x LedgerSummaryItem
|
|
var ms int64
|
|
if err := rows.Scan(&x.Reason, &x.DeltaMicros, &x.Count, &ms, &x.Day); err != nil {
|
|
return nil, err
|
|
}
|
|
x.CreatedAt = time.UnixMilli(ms).UTC()
|
|
out = append(out, x)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
func (s *Store) AddLedgerOnce(ctx context.Context, cid string, delta int64, reason, ref string) (bool, error) {
|
|
if delta == 0 {
|
|
return false, nil
|
|
}
|
|
res, err := s.DB.ExecContext(ctx, `INSERT OR IGNORE INTO credit_ledger(customer_id,delta_micros,reason,reference,created_at) VALUES(?,?,?,?,?)`, cid, delta, reason, ref, time.Now().UTC().UnixMilli())
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
n, err := res.RowsAffected()
|
|
return n == 1, err
|
|
}
|
|
|
|
func (s *Store) SeedSetting(ctx context.Context, key, value string) error {
|
|
_, err := s.DB.ExecContext(ctx, `INSERT OR IGNORE INTO customer_settings(key,value,updated_at) VALUES(?,?,?)`, key, value, time.Now().UTC().UnixMilli())
|
|
return err
|
|
}
|
|
func (s *Store) Setting(ctx context.Context, key, fallback string) string {
|
|
var v string
|
|
if err := s.DB.QueryRowContext(ctx, `SELECT value FROM customer_settings WHERE key=?`, key).Scan(&v); err != nil {
|
|
return fallback
|
|
}
|
|
return v
|
|
}
|
|
func (s *Store) SetSetting(ctx context.Context, key, value string) error {
|
|
_, err := s.DB.ExecContext(ctx, `INSERT INTO customer_settings(key,value,updated_at) VALUES(?,?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value,updated_at=excluded.updated_at`, key, value, time.Now().UTC().UnixMilli())
|
|
return err
|
|
}
|
|
|
|
func (s *Store) CreateRegistrationInvite(ctx context.Context, tokenHash, label string, expiresAt time.Time) error {
|
|
_, err := s.DB.ExecContext(ctx, `INSERT INTO registration_invites(token_hash,label,expires_at,created_at) VALUES(?,?,?,?)`, tokenHash, strings.TrimSpace(label), expiresAt.UTC().UnixMilli(), time.Now().UTC().UnixMilli())
|
|
return err
|
|
}
|
|
func (s *Store) ActiveInviteCount(ctx context.Context) int {
|
|
var n int
|
|
_ = s.DB.QueryRowContext(ctx, `SELECT count(*) FROM registration_invites WHERE used_by IS NULL AND expires_at>?`, time.Now().UTC().UnixMilli()).Scan(&n)
|
|
return n
|
|
}
|
|
|
|
func (s *Store) WorkerByClientID(ctx context.Context, clientID string) (Worker, error) {
|
|
return scanWorker(s.DB.QueryRowContext(ctx, `SELECT `+workerCols+` FROM workers WHERE worker_client_id=? ORDER BY created_at DESC LIMIT 1`, strings.TrimSpace(clientID)))
|
|
}
|
|
|
|
func scanWorker(row interface{ Scan(...any) error }) (Worker, error) {
|
|
var w Worker
|
|
var last, lastLease sql.NullInt64
|
|
var created, updated int64
|
|
err := row.Scan(&w.ID, &w.CustomerID, &w.TaskID, &w.BeaconPath, &w.ContainerID, &w.ControllerID, &w.Volume, &w.Status, &w.WorkerClientID, &w.RegisterToken, &w.RateMicrosPerMinute, &last, &lastLease, &created, &updated, &w.LastError)
|
|
if err != nil {
|
|
return w, err
|
|
}
|
|
if last.Valid {
|
|
v := time.UnixMilli(last.Int64).UTC()
|
|
w.LastChargeAt = &v
|
|
}
|
|
if lastLease.Valid {
|
|
v := time.UnixMilli(lastLease.Int64).UTC()
|
|
w.LastLeaseAt = &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,controller_id,docker_volume,status,worker_client_id,register_token,rate_micros_per_minute,last_charge_at,last_lease_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,controller_id,docker_volume,status,register_token,rate_micros_per_minute,created_at,updated_at) VALUES(?,?,?,?,?,?,'stopped',?,?,?,?)`, w.ID, w.CustomerID, w.TaskID, w.BeaconPath, w.ControllerID, 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) SetWorkerController(ctx context.Context, wid, controllerID string) error {
|
|
_, err := s.DB.ExecContext(ctx, `UPDATE workers SET controller_id=?,updated_at=? WHERE id=?`, strings.TrimSpace(controllerID), time.Now().UTC().UnixMilli(), wid)
|
|
return err
|
|
}
|
|
func (s *Store) TouchWorkerLease(ctx context.Context, wid string) error {
|
|
_, err := s.DB.ExecContext(ctx, `UPDATE workers SET last_lease_at=?,updated_at=? WHERE id=?`, time.Now().UTC().UnixMilli(), 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()
|
|
}
|