Control plane

Function-level reference for main.go, the API server, scheduler, alert manager, usage recorder, and structured logging.

Files: main.go, internal/api/*, internal/scheduler/scheduler.go, internal/alerts/*, internal/usage/recorder.go, internal/applog/*.

main.go

Application entry point. Wires every subsystem in dependency order: loads configuration, opens and configures the database, applies optional plan overrides, constructs the scheduler, log bus, webhook dispatcher, alert manager + threshold poller, usage recorder, API server, and web handler, then blocks on an OS signal before performing an orderly shutdown.

Functions

main() — Bootstraps and runs the entire platform; starts the scheduler, worker-health monitor, API server, alert threshold poller, and usage recorder as goroutines, then waits for SIGINT/SIGTERM. Side effects: spawns goroutines for sched.Start, sched.MonitorWorkerHealth, server.Run, thresholdPoller.Start, usageRecorder.Start, plus a 6-hour ticker that calls db.PruneRevokedSessions(30d) and db.PrunePendingTOTPs(); registers the Hub.PushJob callback with the scheduler; wires the webhook publisher into the threshold poller; defers Stop() on each background subsystem and database close.

internal/api/server.go

Defines the HTTP API server, its Gin router setup, authentication middleware, and the REST endpoints (/health, /v1/workers/register, /v1/stats). Also exposes GenerateAPIKey.

Type

Server — Holds the Gin router, DB, scheduler, config, log bus, and the WS Hub.

Functions

NewServer(db, sched, cfg, bus, publisher) *Server — Constructs a server, sets Gin to release mode, registers a logger that skips /v1/workers/ws and /health, creates the Hub (passing cfg.APIServerSecret() so it can mint/verify worker JWTs), and registers routes.
(s *Server) Hub() *Hub — Returns the WS hub.
(s *Server) setupRoutes() — Registers GET /health, POST /v1/workers/register (controlAuth), POST /v1/workers/refresh-token (controlAuth — Bearer raw API key), GET /v1/workers/ws, GET /v1/traefik/config (worker Traefik bearer token auth — zone-subdomain routing), GET /v1/traefik/proxy-config (proxy bearer token auth — custom-domain routing), GET /v1/stats (controlAuth).
(s *Server) Run(addr string) error — Starts HTTP or TLS 1.3 listener via a stored httpSrv.
(s *Server) Shutdown(ctx context.Context) error — Graceful shutdown of the HTTP server.
(s *Server) Router() *gin.Engine — Returns the engine for external mounting (used by web.Handler).
extractBearer(header) string — Strips "Bearer " prefix.
(s *Server) controlAuth(c *gin.Context) — Constant-time compare of bearer against api_server.secret; aborts with 401 on mismatch.
(s *Server) healthCheck(c) — Returns {"status":"ok","time":...}.
(s *Server) registerWorker(c) — Validates api_key, calls db.UpsertWorker (stores hashed key only), then mints a 15-minute worker JWT via signWorkerJWT. Response: {worker, token, token_expires_at}.
(s *Server) refreshWorkerToken(c) — Bearer is the raw long-lived API key; resolves the worker via the hashed lookup and returns a fresh JWT + expiry. Workers call this ~2 minutes before their current token expires.
(s *Server) getStats(c) — Returns scheduler.GetWorkerStats() aggregate.
GenerateAPIKey(byteLen) (string, error) — Hex-encoded crypto/rand bytes.

internal/api/ws.go

WebSocket hub and per-connection lifecycle for worker nodes. Handles upgrade/auth, maintains the active workerConn map, and provides read/write loops, sequence dedup, log rate-limiting, and outbound job delivery.

Types

WebhookPublisher — interface with Publish(e webhooks.Event).
Hub — registry of live worker connections (conns map[int]*workerConn) under sync.RWMutex; holds a *statsStore.
workerConn — single authenticated WS connection: send channel, done channel, atomic.Bool closed, outbound outSeq, inbound dedup state, log-token-bucket state.

Constants

wsWriteWait 10 s, wsPongWait 90 s, wsPingPeriod 72 s, log ingress 256 KiB/s sustained / 1 MiB burst, max line 8 KiB.

Functions

NewHub(db, sched, bus, publisher, alertMgr, masterSecret) *Hub — Allocates a Hub with a 45-second TTL stats store. masterSecret is api_server.secret; the hub uses it to verify worker JWTs presented at WSS upgrade.
(h *Hub) AppStats(appID) (cpuPct, memBytes, memLimit, updatedAt, ok) — Reads cached snapshot.
(h *Hub) AppStatsLive(appID) (LiveAppStats, ok) — Rich snapshot (CPU%, mem, mem-limit, net Mbps, net total bytes, disk Mbps).
(h *Hub) PushJob(workerID, jobID) — Looks up live conn under RLock, calls sendJob.
(h *Hub) resolveAuthForUpgrade(token) (*Worker, []byte hmacKey, ok bool) — Two-path auth. If token has the JWT shape (two dots), verifyWorkerJWT over masterSecret, then FindWorkerByID(claims.Sub), then a constant-time check that claims.KeyHash == w.APIKeyHash so a leaked JWT becomes useless after key rotation; the HMAC signing key for envelopes is recovered by hex-decoding APIKeyHash (the raw 32-byte sha256 output). Else falls back to the raw API key path: FindOnlineWorkerByAPIKey + HashAPIKeyBytes(token) for the HMAC key.
(h *Hub) handleUpgrade(c) — Reads the bearer token from Authorization. Calls resolveAuthForUpgrade (accepts JWT or raw key). Marks worker online if it was offline, upgrades HTTP to WS, replaces any existing conn for that worker, launches writeLoop, readLoop, replayPendingJobs.
(h *Hub) deregister(wc) — Removes conn from h.conns if it is still current.
(wc *workerConn) close(reason) — Idempotent shutdown via atomic.Bool; cancels ctx, closes done, closes WS, deregisters.
(wc *workerConn) writeLoop() — Drains send, sends periodic pings.
(wc *workerConn) readLoop() — Reads frames, verifies HMAC, validates clock, checks sequence, calls dispatch.
(wc *workerConn) acceptSeq(seq) bool — Sliding-window dedup (256 entries, prune at 1024).
(wc *workerConn) writeEnvelope(env) error — Non-blocking enqueue with 5 s timeout.
(wc *workerConn) emit(msgType, payload) error — Builds signed envelope, enqueues.
(wc *workerConn) sendError(inReplyTo, code, msg) — Emits a TypeError.
(wc *workerConn) sendJob(jobID) error — Loads job from DB, decodes DeployPayload, emits TypeJob.
(wc *workerConn) replayPendingJobs() — Queries NextAssignedJob for the worker and sendJob if any.
stripANSI(s) string — Removes ANSI escapes and null bytes.
capLine(s) string — Truncates a string to 8 KiB.
(wc *workerConn) allowLogBytes(n) bool — Token-bucket rate limiter.

internal/api/ws_dispatch.go

Message-type dispatcher and inbound handler methods for the worker WSS protocol.

(wc) dispatch(env) — Routes envelope by Type; sends unknown_type error otherwise.
(wc) handleAppStats(env) — Decodes AppStatsPayload; verifies app belongs to this worker; stores snapshot in hub.stats.
(wc) handleHeartbeat(env) — Decodes payload; calls db.UpdateWorkerHeartbeat.
(wc) handleJobStatus(env) — Decodes payload; calls db.UpdateJobStatus.
(wc) handleJobResult(env) — Decodes, sanitises logs, calls db.CompleteJob, releases worker resources via scheduler, invalidates app cache, emits webhook event, spawns replayPendingJobs.
(wc) emitJobEvent(job, p, deploy) — Publishes EventDeploymentSucceeded / Failed for deploy / restart only.
(wc) handleAppStatus(env) — On "stopped", marks app stopped + publishes EventAppStopped.
(wc) handleViolation(env) — Marks app violated, invalidates cache, publishes EventAppViolation.
(wc) handleRuntimeLog(env) — Rate-checks, decodes, caps lines at 5000, verifies app ownership per line, sanitises, pushes to log bus.
(wc) handleBuildLog(env) — Rate-checks, decodes, calls db.SetDeploymentBuildLogs.
capLogs(s, max) string — When len(s) > max, keeps the first max/2 and last max/2 bytes joined with a ...[truncated]... marker — preserves both the early failure context and the end of the run rather than only the tail.

internal/api/traefik.go

Traefik HTTP-provider endpoints. Two endpoints exist: workerConfig (zone-subdomain routing per worker) and proxyConfig (custom-domain routing via the proxy server). Both support ETag/If-None-Match — responses include a SHA-256-based ETag, returning 304 Not Modified when the config hasn't changed.

Types

traefikDynamicConfig, traefikHTTP, traefikRouter, traefikService, traefikLB, traefikServer, traefikTLS — JSON shape Traefik consumes on each poll. Only emitted fields are typed.

Functions

(s) workerConfig(c) — Worker Traefik endpoint (GET /v1/traefik/config?worker_id=N). Authenticates via a static bearer token compared against cfg.WorkerTraefikToken() (constant-time). Calls db.ListRunningAppsForWorker(workerID) to enumerate running apps on this worker. For each app with a zone, emits one router with rule Host(\appname.zone`)pointing tohttp://localhost:PORT`, entryPoints: ["web"]. No TLS — zone traffic is fronted by Cloudflare.
(s) proxyConfig(c) — Proxy server endpoint (GET /v1/traefik/proxy-config). Authenticates via a static Bearer token compared against cfg.ProxyToken() (constant-time). Calls db.ListRunningAppsWithWorkers() to enumerate all running apps that have a worker with a non-empty WireGuardIP. Uses db.ListVerifiedDomainsByAppIDs (batch query) to fetch verified custom domains in one round-trip. For each app with custom domains, emits one router with rule Host(\custom.domain`), one service pointing at http://WORKER_WG_IP:CONTAINER_PORT(direct WireGuard upstream),entryPoints: ["web", "websecure"], and tls.certResolver = "letsencrypt"`.
buildHostRule(hosts) string — Renders Host(\a`) || Host(`b`)from a slice of validated hostnames (already passedvalidateHostname` upstream so backtick injection is impossible).
safeRouterName(name) string — Belt-and-suspenders: keeps router/service names ASCII-safe. App slugs already meet the rule.

internal/api/worker_jwt.go

Short-lived bearer tokens scoped to a specific worker. Reduces the blast radius of a leaked Authorization header by bounding useful lifetime to ~15 minutes — far shorter than the long-lived API key that mints them. Signing key is HKDF-derived from the master secret so it rotates whenever the master secret rotates.

Constants

workerJWTTTL = 15 * time.Minute
workerJWTRefreshSkew = 2 * time.Minute — workers refresh this far before expiry
workerJWTKeyLen = 32
workerJWTHKDFPurpose = "wisehosting-worker-jwt-v1"

Type

workerJWTClaims { Sub int (worker ID), KeyHash string (sha256 hex of API key — binds JWT to current key), Iat, Exp int64, Jti string }

Functions

hkdfKey(masterSecret, purpose, n) []byte — Local HKDF-SHA256 deriver with a purpose-specific info label so JWT-signing, AES, and OAuth-state keys never collide.
signWorkerJWT(masterSecret, workerID, apiKeyHash) (token, exp, err) — HS256, header {"alg":"HS256","typ":"JWT"}, base64url-no-pad segments. Random 12-byte JTI.
verifyWorkerJWT(masterSecret, token) (*workerJWTClaims, error) — Splits on ., checks segment count, HMAC-compares signature in constant time, parses claims, rejects on now ≥ exp.

internal/api/stats_store.go

Thread-safe in-memory cache for per-app CPU/memory snapshots with subscription support for push-based updates. Lazy GC on Put.

Type — AppStatSnapshot { CPUPct, MemBytes, MemLimit, UpdatedAt }, LiveAppStats { CPUPct, MemBytes, MemLimit, NetMBps, NetTotalBytes, DiskMBps, UpdatedAt }, statsStore (RWMutex map + subscriber registry).

newStatsStore(ttl) *statsStore — Pre-sized map.
(s) Put(appID, snap) — Writes; sweeps stale entries when TTL has passed since last GC; fans out LiveAppStats to all subscribers for this appID.
(s) Get(appID) (snap, ok) — Returns false if missing or stale.
(s) AppStatsLive(appID) (LiveAppStats, ok) — Rich snapshot including network and disk metrics.
(s) SubscribeStats(appID) (<-chan LiveAppStats, cancel func()) — Returns a channel that receives live stats updates for the given app, plus a cancel function to unsubscribe. Used by WebSocket handlers (wsUsage, wsStats) to push real-time updates to the dashboard.
(s) DeleteApp(appID) — Removes entry.

internal/scheduler/scheduler.go

Job scheduling engine: picks pending jobs, assigns them inside a serialisable transaction to the lowest-utilisation eligible worker, recovers stuck jobs, and monitors worker heartbeats. Failed jobs stay failed — there is no automatic retry path; users redeploy from the UI.

Type

Scheduler — *GormDB, polling interval, cancellable ctx, onJobAssigned callback.

Functions

NewScheduler(db, interval) *Scheduler — Allocates with derived ctx.
(s) SetJobAssignedCallback(fn) — Used by main to wire hub.PushJob.
(s) SelectAvailableWorker(resources) (workerID, zone, err) — Read-only weighted-score query (CPU%, memory%, running-app count); no DB writes.
(s) Start() — Ticks at interval: processPendingJobs, recoverStuckJobs.
(s) Stop() — Cancels context.
(s) processPendingJobs() error — Opens SELECT ... FOR UPDATE SKIP LOCKED tx, picks up to 100 pending jobs ordered by priority/age, assigns or honours pre-set WorkerID. After commit, fires onJobAssigned for each.
(s) selectWorkerTx(tx, required) (id, err) — Same scoring inside an existing tx.
(s) assignJobTx(tx, jobID, workerID, resources) error — Atomically increments used_cpu/used_memory (with capacity guard), sets job assigned.
(s) ReleaseWorkerResources(workerID, resources) error — Decrements counters with GREATEST(..., 0) guards.
(s) recoverStuckJobs() error — Marks assigned/processing jobs older than 15 min as failed. Stays failed — no retry.
(s) MonitorWorkerHealth() — Ticks every minute, calls checkWorkerHealth.
(s) checkWorkerHealth() error — Marks online workers with last_heartbeat older than 2 min as offline; bulk-stops their running apps.
(s) GetWorkerStats() (map, err) — Aggregates total/online/offline workers + total and used CPU/memory.
IsRecordNotFound(err) bool — Wraps errors.Is(err, gorm.ErrRecordNotFound).

internal/alerts/manager.go

Alert manager: emits new alert rows, deduplicates against an active alert with the same (user_id, source_type, source_id), resolves them when the underlying condition clears.

Types

Spec { UserID, AppID, Kind, Severity, Source, SourceID, Title, Message } — input for Emit.
Manager — wraps *GormDB.

Functions

NewManager(db) *Manager.
(m) Emit(s Spec) — UPSERT-like: returns the existing active alert if (user, source_type, source_id) already fires; otherwise inserts a new row.
(m) ResolveBySource(userID int, sourceType, sourceID string) — flips matching active alerts to resolved and stamps resolved_at.

internal/alerts/threshold.go

Per-app threshold poller. Runs every 30s, evaluates each enabled AppAlertRule against the live stats snapshot from the WSS hub, and fires/resolves with a configurable sustain window.

Types

Snapshot { CPUPct, MemBytes, MemLimit, NetMBps, NetTotalBytes, DiskMBps, UpdatedAt } — what the poller needs from each running app.
StatsReader — AppStatsLive(appID) (Snapshot, ok).
ThresholdPoller — wraps DB, stats source, manager, optional webhook publisher.

Functions

NewThresholdPoller(db, stats, mgr) *ThresholdPoller.
(p) SetPublisher(pub) — wires webhook dispatch (optional; alerts still emit rows without it).
(p) Start() / (p) Stop() — goroutine lifecycle.
(p) loop() — 30s ticker.
(p) tick() — loads running apps + their rules, evaluates each.
(p) evaluateMetric(app, rule, val, now) — handles cpu / memory / network / disk with hysteresis (sustain over → fire; sustain under → resolve).
(p) evaluateOffline(app, rule, snap, snapOK, now) — same hysteresis but driven off heartbeat staleness (no metric value).
metricValue(kind, snap, ok) / metricEventName(kind) / metricTitle(kind, app) / metricMessage(kind, val, threshold, sustainMin) — helpers shared by fire paths.

The fire path inserts an alerts row via Manager.Emit and (if a publisher is wired) calls Publish with one of app.cpu, app.memory, app.network, app.disk, app.offline. Event-driven kinds (crashloop, deployment_failed) come from elsewhere — the dispatcher and worker — and bypass this poller.

internal/usage/recorder.go

Background sampler that writes per-app live stats into 5-minute usage_samples buckets, then prunes rows older than 90 days once a day.

Constants

BucketSize = 5 * time.Minute
retention = 90 * 24 * time.Hour

Types

LiveStats — AppStatsLive(appID) (LiveSnapshot, ok) interface.
LiveSnapshot { CPUPct, MemBytes, NetMBps, NetTotalBytes } — per-tick reading.
Recorder — owns the in-memory accumulator, ticker, prune loop.
acc — running totals for the current bucket: avg/max for cpu/mem/net, last-seen NetTotalBytes for delta computation.

Functions

New(db, stats) *Recorder.
(r) Start() — kicks off the 60s sample loop and the daily pruner.
(r) Stop() — cancels both.
(r) sample(now) — pulls every running app from the stats source, updates the accumulator, and on bucket rollover calls commit.
(r) commit(appID, bucket, a) — UPSERTs one usage_samples row.
buildSample(appID, userID, bucket, a) — derives net_bytes_delta from NetTotalBytes change, fills *_avg / *_max columns.
bucketStart(t) — UTC-truncate to 5 minutes.
(r) pruneLoop() / (r) prune() — daily DELETE WHERE bucket_start < now-90d.

internal/webhooks/formatters.go

Renders compact {emoji} {title} — {app}\nhttps://{subdomain} strings for Shoutrrr-style targets where signed JSON is the wrong fit (Discord, Slack, ntfy, …).

formatShoutrrrMessage(ev) string — entry point used by the dispatcher when the URL is a Shoutrrr scheme.
eventEmoji(name) / eventTitle(name) — closed lookup tables.
short(sha) — 7-char SHA prefix.
firstLine(s) — strips multi-line commit messages down to a chat-friendly title.

internal/applog/ring.go

Lock-free-read-friendly ring buffer for structured log entries with subscriber support. Holds up to 10 000 entries in memory and supports real-time push via subscriber channels.

Types

Entry — ID int64, Time, Level, Msg, Source string, Fields map[string]any.
Ring — Fixed-size circular buffer with monotonic sequence IDs and subscriber registry.

Functions

NewRing(maxSize) *Ring — Pre-allocates the ring; defaults to 10 000 if ≤ 0.
(r) Append(e Entry) — Assigns a monotonic ID, writes to the ring, fans out to all subscribers.
(r) Query(sinceID, level, search, component, limit) []Entry — Filtered read with case-insensitive search; max 2 000 entries per call.
(r) Latest(limit) []Entry — Returns the N most recent entries.
(r) Subscribe() (<-chan Entry, cancel func()) — Returns a buffered channel (256 cap) that receives every new entry. Caller must call cancel to avoid leaking.

internal/applog/handler.go

Structured logging setup that routes slog output to both stderr and the in-memory ring buffer. Powers the admin "System Logs" view via the internal-api.

Types

RingHandler — slog.Handler implementation that writes structured entries into a Ring.
MultiHandler — Fans out to multiple slog.Handlers (stderr text + ring buffer).

Functions

NewRingHandler(ring, level) *RingHandler — Constructor.
(h) Handle(ctx, record) error — Extracts caller source (trimmed to internal/ or cmd/), collects attrs/groups, appends to ring.
NewMultiHandler(handlers...) *MultiHandler — Fans out each log record to all handlers.
Setup() *Ring — Initialises structured logging for the whole process. Creates a ring (10 000 entries), configures a TextHandler (stderr, debug level, with source) and a RingHandler, combines them via MultiHandler, sets as the global slog default. Returns the ring for query endpoints.
GinWriter() io.Writer — Adapter that routes Gin's log.Printf calls through slog at INFO level with component: "http".

internal/logbus/bus.go

Per-app runtime log buffer. Holds the last N lines (default 5 000) per app in memory and supports real-time push subscriptions for the WebSocket log stream. Distinct from applog — this is for container stdout/stderr, not control-plane structured logs.

Types

Line — Seq int64, AppID int, Stream string, Msg string, At time.Time.
Bus — Per-app ring buffers + subscriber registry.

Functions

New(maxKeepPerApp) *Bus — Defaults to 5 000 lines per app.
(b) Stop() — No-op (kept for API compatibility).
(b) Push(appID, stream, msg, at) — Assigns monotonic sequence, appends to the per-app buffer, fans out to matching subscribers.
(b) Subscribe(appIDs) (<-chan Line, cancel func()) — Buffered channel (512 cap) filtered to the given app IDs.
(b) Tail(appID, sinceSeq, limit) []Line — Reads from a single app's buffer.
(b) TailMulti(appIDs, sinceSeq, limit) []Line — Reads across multiple apps, sorted by sequence.
(b) Drop(appID) — Evicts the buffer for a deleted app.

On this page