Add admin console access and platform pages (trust, users, keys, workloads, settings)

This commit is contained in:
Emil
2026-08-03 00:11:47 +03:00
parent 7fb1059401
commit e923627ce0
36 changed files with 1562 additions and 86 deletions
+19 -11
View File
@@ -77,6 +77,7 @@ type storageDeps struct {
artifactRepo usecase.ArtifactRepository
uiReadRepo usecase.UIReadRepository
adminReadRepo usecase.AdminReadRepository
settingsRepo usecase.WorkloadSettingsRepository
taskResultRepo usecase.TaskResultRepository
statsRepo interface {
Counts(ctx context.Context) (tasks, jobs, workers map[string]int, err error)
@@ -159,7 +160,7 @@ func runWithConfig(cfg infra.Config) error {
useCases := httptransport.UseCases{
RegisterWorker: usecase.NewRegisterWorker(workerRepo, clk),
CreateJob: usecase.NewCreateJob(jobRepo, taskRepo, tx, clk),
SubmitDataset: usecase.NewSubmitDataset(blobStore, artifactRepo, jobRepo, taskRepo, tx, clk, cfg.DefaultMaxAttempts, catalog),
SubmitDataset: usecase.NewSubmitDataset(blobStore, artifactRepo, jobRepo, taskRepo, tx, clk, cfg.DefaultMaxAttempts, catalog, deps.settingsRepo),
ClaimTask: usecase.NewClaimTask(taskRepo, jobRepo, workerRepo, tx, clk, cfg.LeaseDuration, catalog),
RenewLease: usecase.NewRenewLease(taskRepo, workerRepo, tx, clk, cfg.LeaseDuration),
CompleteTask: usecase.NewCompleteTask(taskRepo, jobRepo, artifactRepo, workerRepo, taskResultRepo, tx, clk, cfg.QuorumSize, catalog),
@@ -173,16 +174,21 @@ func runWithConfig(cfg infra.Config) error {
GetTaskInput: usecase.NewGetTaskInput(taskRepo, artifactRepo, blobStore),
Dashboard: usecase.NewDashboard(uiReadRepo, catalog),
PreviewArtifact: usecase.NewPreviewArtifact(uiReadRepo, blobStore),
Admin: usecase.NewAdmin(deps.adminReadRepo, uiReadRepo, usecase.AdminNodeInfo{
Version: version,
StartedAt: clk.Now(),
Binary: executablePath(),
Addr: cfg.Addr,
DataDir: cfg.StorageDir,
DBEngine: cfg.DatabaseEngine,
PublicURL: cfg.PublicCoordinatorURL,
Userservice: cfg.UserserviceURL,
}, deps.ready, clk.Now),
Admin: usecase.NewAdmin(deps.adminReadRepo, uiReadRepo, workerRepo, deps.settingsRepo, catalog,
usecase.AdminNodeInfo{
Version: version,
StartedAt: clk.Now(),
Binary: executablePath(),
Addr: cfg.Addr,
DataDir: cfg.StorageDir,
DBEngine: cfg.DatabaseEngine,
PublicURL: cfg.PublicCoordinatorURL,
Userservice: cfg.UserserviceURL,
WorkerToken: func() string { return cfg.Token },
}, deps.ready, clk.Now).
WithAuditLog(log, func(ctx context.Context, action, detail string) {
log.Info("admin audit", "action", action, "detail", detail)
}),
}
// Background reapers are tracked so shutdown can wait for them. Without this
@@ -264,6 +270,7 @@ func openSQLite(ctx context.Context, cfg infra.Config, log *slog.Logger) (*stora
artifactRepo: sqlite.NewArtifactRepo(db),
uiReadRepo: sqlite.NewUIReadRepo(db),
adminReadRepo: sqlite.NewAdminReadRepo(db),
settingsRepo: sqlite.NewWorkloadSettingsRepo(db),
taskResultRepo: sqlite.NewTaskResultRepo(db),
statsRepo: sqlite.NewStatsRepo(db),
ready: func(ctx context.Context) error { return db.PingContext(ctx) },
@@ -287,6 +294,7 @@ func openPostgres(ctx context.Context, cfg infra.Config, log *slog.Logger) (*sto
artifactRepo: postgres.NewArtifactRepo(pool),
uiReadRepo: postgres.NewUIReadRepo(pool),
adminReadRepo: postgres.NewAdminReadRepo(pool),
settingsRepo: postgres.NewWorkloadSettingsRepo(pool),
taskResultRepo: postgres.NewTaskResultRepo(pool),
statsRepo: postgres.NewStatsRepo(pool),
ready: func(ctx context.Context) error { return pool.Ping(ctx) },
+1
View File
@@ -18,4 +18,5 @@ var (
ErrResultConflict = errors.New("different result already recorded")
ErrInvalidInput = errors.New("invalid input")
ErrTaskNotLeased = errors.New("task is not currently leased")
ErrWorkloadDisabled = errors.New("workload is disabled")
)
+11
View File
@@ -314,6 +314,17 @@ func (r *WorkerRepo) MarkStaleOffline(ctx context.Context, cutoff time.Time) (in
return n, nil
}
func (r *WorkerRepo) SetTrust(ctx context.Context, id uuid.UUID, trust domain.WorkerTrust) error {
r.mu.Lock()
defer r.mu.Unlock()
w, ok := r.workers[id]
if !ok {
return domain.ErrWorkerNotFound
}
w.TrustLevel = trust
return nil
}
// --- ArtifactRepo --------------------------------------------------------
type ArtifactRepo struct {
@@ -54,6 +54,8 @@ func expectedMigrationName(version int) string {
return "0012_worker_trust.up.sql"
case 13:
return "0013_task_results.up.sql"
case 14:
return "0014_workload_settings.up.sql"
default:
return ""
}
@@ -0,0 +1,5 @@
BEGIN;
DROP TABLE IF EXISTS workload_settings;
COMMIT;
@@ -0,0 +1,11 @@
BEGIN;
-- Per-workload enable/disable. Absence of a row means "enabled" (the catalog
-- default); a row only exists once an admin flipped a workload off or back on.
CREATE TABLE workload_settings (
workload text NOT NULL PRIMARY KEY,
enabled boolean NOT NULL,
updated_at timestamptz NOT NULL DEFAULT now()
);
COMMIT;
@@ -91,6 +91,24 @@ func (r *WorkerRepo) MarkStaleOffline(ctx context.Context, cutoff time.Time) (in
return tag.RowsAffected(), nil
}
func (r *WorkerRepo) SetTrust(ctx context.Context, id uuid.UUID, trust domain.WorkerTrust) error {
sql, args, err := psql.Update("workers").
SetMap(map[string]any{"trust_level": string(trust), "updated_at": time.Now()}).
Where(sq.Eq{"id": id}).
ToSql()
if err != nil {
return err
}
tag, err := conn(ctx, r.pool).Exec(ctx, sql, args...)
if err != nil {
return fmt.Errorf("set worker trust: %w", err)
}
if tag.RowsAffected() == 0 {
return domain.ErrWorkerNotFound
}
return nil
}
func scanWorker(row pgx.Row) (*domain.Worker, error) {
var (
w domain.Worker
@@ -0,0 +1,67 @@
package postgres
import (
"context"
"fmt"
"time"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/emil28092005/SciMesh/coordinator/internal/usecase"
)
// WorkloadSettingsRepo persists the per-workload enable/disable overrides.
// Absence of a row means the workload is enabled (the catalog default).
type WorkloadSettingsRepo struct{ pool *pgxpool.Pool }
func NewWorkloadSettingsRepo(pool *pgxpool.Pool) *WorkloadSettingsRepo {
return &WorkloadSettingsRepo{pool: pool}
}
var _ usecase.WorkloadSettingsRepository = (*WorkloadSettingsRepo)(nil)
func (r *WorkloadSettingsRepo) GetEnabled(ctx context.Context, workload string) (bool, error) {
var enabled bool
err := conn(ctx, r.pool).QueryRow(ctx,
"SELECT enabled FROM workload_settings WHERE workload = $1", workload).Scan(&enabled)
if err != nil && err.Error() == "no rows in result set" {
return true, nil // no override: catalog default enabled
}
if err != nil {
return false, fmt.Errorf("get workload setting: %w", err)
}
return enabled, nil
}
func (r *WorkloadSettingsRepo) List(ctx context.Context) ([]usecase.WorkloadSetting, error) {
rows, err := conn(ctx, r.pool).Query(ctx,
"SELECT workload, enabled, updated_at FROM workload_settings ORDER BY workload ASC")
if err != nil {
return nil, fmt.Errorf("list workload settings: %w", err)
}
defer rows.Close()
var out []usecase.WorkloadSetting
for rows.Next() {
var s usecase.WorkloadSetting
if err := rows.Scan(&s.Workload, &s.Enabled, &s.UpdatedAt); err != nil {
return nil, err
}
out = append(out, s)
}
return out, rows.Err()
}
func (r *WorkloadSettingsRepo) SetEnabled(ctx context.Context, workload string, enabled bool, now time.Time) error {
sql, args, err := psql.Insert("workload_settings").
Columns("workload", "enabled", "updated_at").
Values(workload, enabled, now).
Suffix(`ON CONFLICT (workload) DO UPDATE SET enabled = EXCLUDED.enabled, updated_at = EXCLUDED.updated_at`).
ToSql()
if err != nil {
return err
}
if _, err := conn(ctx, r.pool).Exec(ctx, sql, args...); err != nil {
return fmt.Errorf("set workload setting: %w", err)
}
return nil
}
@@ -0,0 +1,89 @@
package sqlite
import (
"context"
"testing"
"time"
"github.com/google/uuid"
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
)
func TestWorkloadSettingsRepoRoundTrip(t *testing.T) {
db := newTestDB(t)
ctx := context.Background()
repo := NewWorkloadSettingsRepo(db)
// No override: enabled by default.
enabled, err := repo.GetEnabled(ctx, "similarity-search")
if err != nil {
t.Fatal(err)
}
if !enabled {
t.Error("workload without an override must be enabled")
}
now := time.Date(2026, 8, 2, 12, 0, 0, 0, time.UTC)
if err := repo.SetEnabled(ctx, "similarity-search", false, now); err != nil {
t.Fatal(err)
}
enabled, err = repo.GetEnabled(ctx, "similarity-search")
if err != nil {
t.Fatal(err)
}
if enabled {
t.Error("workload must be disabled after the override")
}
// Upsert flips it back and updates the timestamp.
later := now.Add(time.Hour)
if err := repo.SetEnabled(ctx, "similarity-search", true, later); err != nil {
t.Fatal(err)
}
enabled, err = repo.GetEnabled(ctx, "similarity-search")
if err != nil {
t.Fatal(err)
}
if !enabled {
t.Error("workload must be re-enabled after the upsert")
}
list, err := repo.List(ctx)
if err != nil {
t.Fatal(err)
}
if len(list) != 1 || list[0].Workload != "similarity-search" || !list[0].Enabled {
t.Errorf("list = %+v, want the single re-enabled override", list)
}
}
func TestWorkerSetTrust(t *testing.T) {
db := newTestDB(t)
ctx := context.Background()
repo := NewWorkerRepo(db)
worker, err := domain.NewWorker("lab-node", []string{"similarity-search"}, fixedTime())
if err != nil {
t.Fatal(err)
}
if err := repo.Insert(ctx, worker); err != nil {
t.Fatal(err)
}
if err := repo.SetTrust(ctx, worker.ID, domain.WorkerUntrusted); err != nil {
t.Fatal(err)
}
got, err := repo.Get(ctx, worker.ID)
if err != nil {
t.Fatal(err)
}
if got.TrustLevel != domain.WorkerUntrusted {
t.Errorf("trust = %q, want untrusted", got.TrustLevel)
}
if err := repo.SetTrust(ctx, worker.ID, domain.WorkerTrusted); err != nil {
t.Fatal(err)
}
if err := repo.SetTrust(ctx, uuid.New(), domain.WorkerTrusted); err != domain.ErrWorkerNotFound {
t.Errorf("unknown worker trust err = %v, want ErrWorkerNotFound", err)
}
}
@@ -0,0 +1,8 @@
-- 0002: per-workload enable/disable. Absence of a row means "enabled" (the
-- catalog default); a row only exists once an admin flipped a workload off or
-- back on.
CREATE TABLE IF NOT EXISTS workload_settings (
workload TEXT NOT NULL PRIMARY KEY,
enabled INTEGER NOT NULL CHECK (enabled IN (0, 1)),
updated_at INTEGER NOT NULL
);
@@ -69,8 +69,8 @@ func TestMigrateIsIdempotent(t *testing.T) {
if err := db.QueryRowContext(ctx, "PRAGMA user_version").Scan(&version); err != nil {
t.Fatal(err)
}
if version != 1 {
t.Errorf("user_version = %d, want 1", version)
if version != 2 {
t.Errorf("user_version = %d, want 2", version)
}
}
@@ -90,3 +90,20 @@ func (r *WorkerRepo) MarkStaleOffline(ctx context.Context, cutoff time.Time) (in
}
return res.RowsAffected()
}
func (r *WorkerRepo) SetTrust(ctx context.Context, id uuid.UUID, trust domain.WorkerTrust) error {
res, err := conn(ctx, r.db).ExecContext(ctx,
"UPDATE workers SET trust_level = ?, updated_at = ? WHERE id = ?",
string(trust), encodeTime(time.Now()), id.String())
if err != nil {
return err
}
affected, err := res.RowsAffected()
if err != nil {
return err
}
if affected == 0 {
return domain.ErrWorkerNotFound
}
return nil
}
@@ -0,0 +1,75 @@
package sqlite
import (
"context"
"database/sql"
"fmt"
"time"
"github.com/emil28092005/SciMesh/coordinator/internal/usecase"
)
// WorkloadSettingsRepo persists the per-workload enable/disable overrides.
// Absence of a row means the workload is enabled (the catalog default).
type WorkloadSettingsRepo struct{ db *sql.DB }
func NewWorkloadSettingsRepo(db *sql.DB) *WorkloadSettingsRepo { return &WorkloadSettingsRepo{db: db} }
var _ usecase.WorkloadSettingsRepository = (*WorkloadSettingsRepo)(nil)
func (r *WorkloadSettingsRepo) GetEnabled(ctx context.Context, workload string) (bool, error) {
var enabled int
err := conn(ctx, r.db).QueryRowContext(ctx,
"SELECT enabled FROM workload_settings WHERE workload = ?", workload).Scan(&enabled)
if err == sql.ErrNoRows {
return true, nil // no override: catalog default enabled
}
if err != nil {
return false, fmt.Errorf("get workload setting: %w", err)
}
return enabled == 1, nil
}
func (r *WorkloadSettingsRepo) List(ctx context.Context) ([]usecase.WorkloadSetting, error) {
rows, err := conn(ctx, r.db).QueryContext(ctx,
"SELECT workload, enabled, updated_at FROM workload_settings ORDER BY workload ASC")
if err != nil {
return nil, fmt.Errorf("list workload settings: %w", err)
}
defer func() { _ = rows.Close() }()
var out []usecase.WorkloadSetting
for rows.Next() {
var (
name string
enabled int
updatedAt int64
)
if err := rows.Scan(&name, &enabled, &updatedAt); err != nil {
return nil, err
}
out = append(out, usecase.WorkloadSetting{
Workload: name,
Enabled: enabled == 1,
UpdatedAt: decodeTime(updatedAt),
})
}
return out, rows.Err()
}
func (r *WorkloadSettingsRepo) SetEnabled(ctx context.Context, workload string, enabled bool, now time.Time) error {
_, err := conn(ctx, r.db).ExecContext(ctx, `
INSERT INTO workload_settings (workload, enabled, updated_at) VALUES (?, ?, ?)
ON CONFLICT (workload) DO UPDATE SET enabled = excluded.enabled, updated_at = excluded.updated_at`,
workload, boolInt(enabled), now.UnixNano())
if err != nil {
return fmt.Errorf("set workload setting: %w", err)
}
return nil
}
func boolInt(b bool) int {
if b {
return 1
}
return 0
}
@@ -42,7 +42,7 @@ func (s *Server) writeError(w http.ResponseWriter, r *http.Request, err error) {
status := http.StatusInternalServerError
switch {
case errors.Is(err, domain.ErrInvalidInput):
case errors.Is(err, domain.ErrInvalidInput), errors.Is(err, domain.ErrWorkloadDisabled):
status = http.StatusBadRequest
case errors.Is(err, domain.ErrJobNotFound), errors.Is(err, domain.ErrTaskNotFound),
errors.Is(err, domain.ErrWorkerNotFound), errors.Is(err, domain.ErrArtifactNotFound):
@@ -182,6 +182,16 @@ func (s *Server) Handler(token string, uiToken ...string) http.Handler {
ui.Handle("GET /ui/admin/api/system", chain(http.HandlerFunc(s.handleUIAdminSystemJSON), gate, requireAdmin))
ui.Handle("GET /ui/admin/api/jobs", chain(http.HandlerFunc(s.handleUIAdminJobsJSON), gate, requireAdmin))
ui.Handle("GET /ui/admin/api/metrics", chain(http.HandlerFunc(s.handleUIAdminMetricsJSON), gate, requireAdmin))
ui.Handle("GET /ui/admin/api/workers", chain(http.HandlerFunc(s.handleUIAdminWorkersJSON), gate, requireAdmin))
ui.Handle("POST /ui/admin/api/workers/{id}/trust", chain(http.HandlerFunc(s.handleUIAdminSetTrustJSON), gate, requireAdmin))
ui.Handle("GET /ui/admin/api/users", chain(http.HandlerFunc(s.handleUIAdminUsersJSON), gate, requireAdmin))
ui.Handle("POST /ui/admin/api/users/{id}/role", chain(http.HandlerFunc(s.handleUIAdminSetUserRoleJSON), gate, requireAdmin))
ui.Handle("GET /ui/admin/api/worker-keys", chain(http.HandlerFunc(s.handleUIAdminWorkerKeysJSON), gate, requireAdmin))
ui.Handle("POST /ui/admin/api/worker-keys/{id}/revoke", chain(http.HandlerFunc(s.handleUIAdminRevokeKeyJSON), gate, requireAdmin))
ui.Handle("GET /ui/admin/api/workloads", chain(http.HandlerFunc(s.handleUIAdminWorkloadsJSON), gate, requireAdmin))
ui.Handle("POST /ui/admin/api/workloads/{name}/enabled", chain(http.HandlerFunc(s.handleUIAdminSetWorkloadEnabledJSON), gate, requireAdmin))
ui.Handle("GET /ui/admin/api/settings", chain(http.HandlerFunc(s.handleUIAdminSettingsJSON), gate, requireAdmin))
ui.Handle("POST /ui/admin/api/token/reveal", chain(http.HandlerFunc(s.handleUIAdminRevealTokenJSON), gate, requireAdmin))
} else {
for _, rt := range app {
ui.HandleFunc(rt.pattern, rt.handler)
@@ -39,6 +39,7 @@ func newEnvWithUIToken(t *testing.T, ready func(context.Context) error, configur
work := memstore.NewWorkerRepo()
arts := memstore.NewArtifactRepo()
blobs := memstore.NewBlobStore()
settings := memstoreSettings{}
clk := memstore.NewClock(time.Date(2026, 7, 21, 12, 0, 0, 0, time.UTC))
tx := memstore.Tx{}
lease := 2 * time.Minute
@@ -47,7 +48,7 @@ func newEnvWithUIToken(t *testing.T, ready func(context.Context) error, configur
uc := coordhttp.UseCases{
RegisterWorker: usecase.NewRegisterWorker(work, clk),
CreateJob: usecase.NewCreateJob(jobs, tasks, tx, clk),
SubmitDataset: usecase.NewSubmitDataset(blobs, arts, jobs, tasks, tx, clk, 3, testCatalog()),
SubmitDataset: usecase.NewSubmitDataset(blobs, arts, jobs, tasks, tx, clk, 3, testCatalog(), settings),
ClaimTask: usecase.NewClaimTask(tasks, jobs, work, tx, clk, lease, testCatalog()),
RenewLease: usecase.NewRenewLease(tasks, work, tx, clk, lease),
CompleteTask: usecase.NewCompleteTask(tasks, jobs, arts, work, memstore.NewTaskResultRepo(), tx, clk, 2, testCatalog()),
@@ -739,3 +740,22 @@ func (e *env) uploadDataset(t *testing.T, workload string, rows int, tsv string)
}
func itoa(n int) string { return strconv.Itoa(n) }
// memstoreSettings is an in-memory WorkloadSettingsRepository for tests.
type memstoreSettings struct{ overrides map[string]bool }
func (m memstoreSettings) GetEnabled(ctx context.Context, workload string) (bool, error) {
if enabled, ok := m.overrides[workload]; ok {
return enabled, nil
}
return true, nil
}
func (m memstoreSettings) List(ctx context.Context) ([]usecase.WorkloadSetting, error) {
return nil, nil
}
func (m memstoreSettings) SetEnabled(ctx context.Context, workload string, enabled bool, now time.Time) error {
m.overrides[workload] = enabled
return nil
}
@@ -205,13 +205,35 @@ tbody tr:hover{background:var(--panel-2)}
</div>
</section>
<!-- ═══ WORKERS (M2) ═══ -->
<!-- ═══ WORKERS ═══ -->
<section class="page" id="page-workers">
<div class="card"><div class="card-head"><h3>Workers</h3><span>milestone M2</span></div><div class="placeholder"><b>Trust management lands with milestone M2</b>Worker list, trust dropdown and heartbeat overview are wired next. The dashboard already shows the live fleet.</div></div>
<div class="section-note">Workers register themselves. Trust decides whether a machine's results are accepted directly or need quorum.</div>
<div class="card">
<table>
<thead><tr><th>Worker</th><th>Status</th><th>Capabilities</th><th>Trust</th><th>Owner</th><th>Last signal</th></tr></thead>
<tbody id="worker-rows"></tbody>
</table>
</div>
</section>
<!-- ═══ USERS & KEYS (M2) ═══ -->
<!-- ═══ USERS & KEYS ═══ -->
<section class="page" id="page-users">
<div class="section-title">Users</div>
<div class="card">
<table>
<thead><tr><th>Email</th><th>Role</th><th>Verified</th><th>Created</th></tr></thead>
<tbody id="user-rows"></tbody>
</table>
<div class="footer-row"><span id="user-count"></span></div>
</div>
<div class="section-title">Worker keys</div>
<div class="section-note">Keys let lab machines register as workers under a user account. Served instances can also use the cluster token.</div>
<div class="card">
<table>
<thead><tr><th>Label</th><th>Prefix</th><th>Owner</th><th>Created</th><th>Last used</th><th></th></tr></thead>
<tbody id="key-rows"></tbody>
</table>
</div>
<div class="section-title">Quick user action</div>
<div class="card">
<form method="post" action="/ui/admin/user-action" style="padding:16px 20px">
@@ -226,13 +248,17 @@ tbody tr:hover{background:var(--panel-2)}
{{if .Error}}<div style="margin-left:210px;margin-top:10px;color:var(--red);font-size:13px">✗ {{.Error}}</div>{{end}}
</form>
</div>
<div class="section-title">Accounts and worker keys</div>
<div class="card"><div class="placeholder"><b>User list and key management land with milestone M2</b>The userservice gains admin list endpoints; the console gets tables, role selects and key revoke.</div></div>
</section>
<!-- ═══ WORKLOADS (M2) ═══ -->
<!-- ═══ WORKLOADS ═══ -->
<section class="page" id="page-workloads">
<div class="card"><div class="placeholder"><b>Workload enable/disable lands with milestone M2</b>The catalog is already served to the job form; persisted on/off switches arrive with the settings migration.</div></div>
<div class="section-note">Disabled workloads are rejected at submit time and hidden from the job form. Settings persist in the database.</div>
<div class="card">
<table>
<thead><tr><th>Workload</th><th>Reduction</th><th>Parameters</th><th>Dataset upload</th><th>Enabled</th></tr></thead>
<tbody id="workload-rows"></tbody>
</table>
</div>
</section>
<!-- ═══ METRICS ═══ -->
@@ -255,9 +281,20 @@ tbody tr:hover{background:var(--panel-2)}
</div>
</section>
<!-- ═══ SETTINGS (M2) ═══ -->
<!-- ═══ SETTINGS ═══ -->
<section class="page" id="page-settings">
<div class="card"><div class="placeholder"><b>Cluster settings land with milestone M2</b>Worker token reveal, public URL and storage settings follow in the next milestone.</div></div>
<div class="warn-strip" style="display:flex;gap:10px;align-items:flex-start;background:var(--amber-soft);border:1px solid #e5b64f33;border-radius:10px;padding:12px 14px;font-size:12.5px;color:#eecf8d"><span>The cluster token below authenticates <b>any</b> worker. Reveal it only on a trusted machine.</span></div>
<div class="section-title">Cluster</div>
<div class="card">
<dl class="kv">
<dt>Worker token</dt><dd><div class="secret"><code id="tok">••••••••••••••••••••••••</code><button class="btn btn-ghost btn-sm" id="reveal">Reveal</button></div></dd>
<dt>Public URL</dt><dd><code id="s-public"></code></dd>
<dt>Listen address</dt><dd><code id="s-addr"></code></dd>
<dt>Data directory</dt><dd><code id="s-datadir"></code></dd>
<dt>Database engine</dt><dd id="s-engine"></dd>
<dt>Binary</dt><dd><code id="s-binary"></code></dd>
</dl>
</div>
</section>
</div>
@@ -272,10 +309,10 @@ const fmtBytes=b=>{if(b==null||b<0)return '—';if(b<1024)return b+' B';if(b<104
const fmtTime=t=>t?new Date(t).toLocaleString():'—';
const esc=s=>String(s).replace(/[&<>"]/g,c=>({'&':'&amp;','<':'&lt;','>':'&gt;','"':'&quot;'}[c]));
let timer=null,current={page:'system',jobsStatus:'',jobsPage:1};
const setPage=page=>{current.page=page;document.querySelectorAll('.nav-item').forEach(b=>b.classList.toggle('active',b.dataset.page===page));document.querySelectorAll('.page').forEach(p=>p.classList.toggle('active',p.id==='page-'+page));const [t,s]=titles[page];document.getElementById('page-title').textContent=t;document.getElementById('page-sub').textContent=s;if(timer){clearInterval(timer);timer=null}if(page==='system'||page==='jobs'||page==='metrics'){refresh();timer=setInterval(refresh,5000)}};
const setPage=page=>{current.page=page;document.querySelectorAll('.nav-item').forEach(b=>b.classList.toggle('active',b.dataset.page===page));document.querySelectorAll('.page').forEach(p=>p.classList.toggle('active',p.id==='page-'+page));const [t,s]=titles[page];document.getElementById('page-title').textContent=t;document.getElementById('page-sub').textContent=s;if(timer){clearInterval(timer);timer=null}refresh();timer=setInterval(refresh,5000)};
document.querySelectorAll('.nav-item').forEach(b=>b.addEventListener('click',()=>setPage(b.dataset.page)));
const refresh=()=>{if(document.hidden)return;const p=current.page;if(p==='system')loadSystem();else if(p==='jobs')loadJobs();else if(p==='metrics')loadMetrics()};
const refresh=()=>{if(document.hidden)return;const p=current.page;if(p==='system')loadSystem();else if(p==='jobs')loadJobs();else if(p==='metrics')loadMetrics();else if(p==='workers')loadWorkers();else if(p==='users')loadUsers();else if(p==='workloads')loadWorkloads();else if(p==='settings')loadSettings()};
document.addEventListener('visibilitychange',()=>{if(!document.hidden)refresh()});
async function loadSystem(){
@@ -384,6 +421,114 @@ async function loadMetrics(){
}
}
setPage('system');
const workerStatusPill=s=>({online:['Online','pill-success'],busy:['Busy','pill-active'],offline:['Offline','pill-waiting']}[s]||[s,'pill-waiting']);
async function loadWorkers(){
const r=await fetch('/ui/admin/api/workers',{headers:{Accept:'application/json'}});
if(!r.ok)return;
const v=await r.json();
const rows=document.getElementById('worker-rows');
rows.replaceChildren();
if(!v.workers.length){const tr=document.createElement('tr');tr.innerHTML='<td colspan="6"><div class="empty">No worker is registered yet.</div></td>';rows.append(tr);return}
for(const w of v.workers){
const tr=document.createElement('tr');
const [label,cls]=workerStatusPill(w.status);
const trustSel='<select data-id="'+w.id+'" class="trust-sel" '+(w.status==='offline'?'disabled':'')+'><option value="trusted" '+(w.trust==='trusted'?'selected':'')+'>Trusted</option><option value="untrusted" '+(w.trust==='untrusted'?'selected':'')+'>Untrusted</option></select>';
tr.innerHTML='<td><div class="t-main">'+esc(w.name)+'</div><div class="t-sub">'+esc(w.id.slice(0,8))+'…</div></td>'+
'<td>'+pill(label,cls,null)+'</td>'+
'<td>'+(w.capabilities||[]).map(c=>'<span class="cap">'+esc(c)+'</span>').join('')+'</td>'+
'<td>'+trustSel+'</td>'+
'<td style="color:var(--text-2)">'+esc(w.owner)+'</td>'+
'<td style="color:var(--text-2);font-size:12.5px">'+fmtTime(w.last_heartbeat_at)+'</td>';
rows.append(tr);
}
document.querySelectorAll('.trust-sel').forEach(sel=>sel.addEventListener('change',async()=>{
await fetch('/ui/admin/api/workers/'+sel.dataset.id+'/trust',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({trusted:sel.value==='trusted'})});
loadWorkers();
}));
}
async function loadUsers(){
const r=await fetch('/ui/admin/api/users',{headers:{Accept:'application/json'}});
if(!r.ok)return;
const v=await r.json();
const rows=document.getElementById('user-rows');
rows.replaceChildren();
for(const u of v.users||[]){
const tr=document.createElement('tr');
const roleSel='<select class="role-sel" data-id="'+u.id+'"><option value="user" '+(u.role==='user'?'selected':'')+'>user</option><option value="admin" '+(u.role==='admin'?'selected':'')+'>admin</option></select>';
tr.innerHTML='<td><div class="t-main">'+esc(u.email)+'</div></td>'+
'<td>'+roleSel+'</td>'+
'<td>'+pill(u.verified?'Verified':'—',u.verified?'pill-success':'pill-waiting',null)+'</td>'+
'<td style="color:var(--text-2);font-size:12.5px">'+fmtTime(u.created_at)+'</td>';
rows.append(tr);
}
document.getElementById('user-count').textContent=(v.users||[]).length+' users';
document.querySelectorAll('.role-sel').forEach(sel=>sel.addEventListener('change',async()=>{
await fetch('/ui/admin/api/users/'+sel.dataset.id+'/role',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({role:sel.value})});
loadUsers();
}));
const keys=await (await fetch('/ui/admin/api/worker-keys',{headers:{Accept:'application/json'}})).json();
const keyRows=document.getElementById('key-rows');
keyRows.replaceChildren();
const emailOf={};for(const u of v.users||[])emailOf[u.id]=u.email;
for(const k of keys.worker_keys||[]){
const tr=document.createElement('tr');
tr.innerHTML='<td class="t-main">'+esc(k.name)+'</td>'+
'<td><code style="color:var(--text-2)">'+esc(k.prefix)+'…</code></td>'+
'<td style="color:var(--text-2)">'+esc(emailOf[k.user_id]||k.user_id.slice(0,8)+'…')+'</td>'+
'<td style="color:var(--text-2);font-size:12.5px">'+fmtTime(k.created_at)+'</td>'+
'<td style="color:var(--text-2);font-size:12.5px">'+(k.last_used_at?fmtTime(k.last_used_at):'never')+'</td>'+
'<td>'+((k.revoked_at)?'<span class="pill pill-danger"><i></i>Revoked</span>':'<button class="btn btn-danger btn-sm key-revoke" data-id="'+k.id+'">Revoke</button>')+'</td>';
keyRows.append(tr);
}
document.querySelectorAll('.key-revoke').forEach(btn=>btn.addEventListener('click',async()=>{
if(!confirm('Revoke this worker key? The machine will be cut off on its next refresh.'))return;
await fetch('/ui/admin/api/worker-keys/'+btn.dataset.id+'/revoke',{method:'POST'});
loadUsers();
}));
}
async function loadWorkloads(){
const r=await fetch('/ui/admin/api/workloads',{headers:{Accept:'application/json'}});
if(!r.ok)return;
const v=await r.json();
const rows=document.getElementById('workload-rows');
rows.replaceChildren();
for(const w of v.workloads||[]){
const tr=document.createElement('tr');
tr.innerHTML='<td><div class="t-main">'+esc(w.name)+'</div><div class="t-sub" style="font-family:inherit">'+esc(w.description||'')+'</div></td>'+
'<td><span class="pill pill-active"><i></i>'+esc(w.reduction)+'</span></td>'+
'<td style="color:var(--text-2);font-size:12.5px">'+w.parameters+' declared</td>'+
'<td>'+pill(w.upload_ready?'ready':'—',w.upload_ready?'pill-success':'pill-waiting',null)+'</td>'+
'<td><button class="toggle wl-toggle '+(w.enabled?'on':'')+'" data-name="'+w.name+'" aria-label="enabled"></button></td>';
rows.append(tr);
}
document.querySelectorAll('.wl-toggle').forEach(t=>t.addEventListener('click',async()=>{
const enabled=!t.classList.contains('on');
await fetch('/ui/admin/api/workloads/'+encodeURIComponent(t.dataset.name)+'/enabled',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({enabled})});
t.classList.toggle('on',enabled);
}));
}
async function loadSettings(){
const r=await fetch('/ui/admin/api/settings',{headers:{Accept:'application/json'}});
if(!r.ok)return;
const v=await r.json();
document.getElementById('s-public').textContent=v.public_url||'—';
document.getElementById('s-addr').textContent=v.addr;
document.getElementById('s-datadir').textContent=v.data_dir||'—';
document.getElementById('s-engine').textContent=v.db_engine;
document.getElementById('s-binary').textContent=v.binary||'—';
}
document.getElementById('reveal').addEventListener('click',async e=>{
const tok=document.getElementById('tok');
if(tok.textContent.startsWith('•')){
const r=await fetch('/ui/admin/api/token/reveal',{method:'POST',headers:{Accept:'application/json'}});
if(!r.ok)return;
const v=await r.json();
tok.textContent=v.token||'(none)';
e.target.textContent='Hide';
}else{tok.textContent='••••••••••••••••••••••••';e.target.textContent='Reveal'}
});
document.querySelectorAll('.toggle').forEach(t=>t.addEventListener('click',()=>t.classList.toggle('on')));
</script>
</body>
</html>
@@ -12,6 +12,7 @@ import (
"github.com/google/uuid"
"github.com/emil28092005/SciMesh/coordinator/internal/authctx"
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
)
// adminUserActions are the userservice endpoints the admin panel may invoke, by
@@ -157,6 +158,199 @@ func (s *Server) handleUIAdminMetricsJSON(w http.ResponseWriter, r *http.Request
writeJSON(w, http.StatusOK, view)
}
// handleUIAdminWorkersJSON serves the admin "Workers" page.
func (s *Server) handleUIAdminWorkersJSON(w http.ResponseWriter, r *http.Request) {
ctx, cancel := s.reqCtx(r)
defer cancel()
view, err := s.uc.Admin.Workers(ctx, s.adminOwnerEmails(r))
if err != nil {
s.writeError(w, r, err)
return
}
writeJSON(w, http.StatusOK, view)
}
// handleUIAdminSetTrustJSON flips one worker's trust level.
func (s *Server) handleUIAdminSetTrustJSON(w http.ResponseWriter, r *http.Request) {
id, err := uuid.Parse(r.PathValue("id"))
if err != nil {
s.writeError(w, r, domain.ErrInvalidInput)
return
}
var body struct {
Trusted bool `json:"trusted"`
}
if err := decodeJSON(r, &body); err != nil {
s.writeError(w, r, domain.ErrInvalidInput)
return
}
ctx, cancel := s.reqCtx(r)
defer cancel()
if err := s.uc.Admin.SetTrust(ctx, id, body.Trusted); err != nil {
s.writeError(w, r, err)
return
}
w.WriteHeader(http.StatusNoContent)
}
// handleUIAdminWorkloadsJSON serves the catalog with persisted enable flags.
func (s *Server) handleUIAdminWorkloadsJSON(w http.ResponseWriter, r *http.Request) {
ctx, cancel := s.reqCtx(r)
defer cancel()
view, err := s.uc.Admin.Workloads(ctx)
if err != nil {
s.writeError(w, r, err)
return
}
writeJSON(w, http.StatusOK, view)
}
// handleUIAdminSetWorkloadEnabledJSON flips a workload's enable flag.
func (s *Server) handleUIAdminSetWorkloadEnabledJSON(w http.ResponseWriter, r *http.Request) {
name := r.PathValue("name")
var body struct {
Enabled bool `json:"enabled"`
}
if err := decodeJSON(r, &body); err != nil {
s.writeError(w, r, domain.ErrInvalidInput)
return
}
ctx, cancel := s.reqCtx(r)
defer cancel()
if err := s.uc.Admin.SetWorkloadEnabled(ctx, name, body.Enabled); err != nil {
s.writeError(w, r, err)
return
}
w.WriteHeader(http.StatusNoContent)
}
// handleUIAdminSettingsJSON serves the read-only cluster settings.
func (s *Server) handleUIAdminSettingsJSON(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, s.uc.Admin.Settings())
}
// handleUIAdminRevealTokenJSON reveals the shared worker token, auditing the
// reveal. Admin-only via the route chain.
func (s *Server) handleUIAdminRevealTokenJSON(w http.ResponseWriter, r *http.Request) {
actor := "admin"
if req, ok := authctx.From(r.Context()); ok {
actor = req.Role + ":" + req.UserID.String()
}
ctx, cancel := s.reqCtx(r)
defer cancel()
writeJSON(w, http.StatusOK, map[string]string{"token": s.uc.Admin.RevealWorkerToken(ctx, actor)})
}
// handleUIAdminUsersJSON serves the account table, proxied from the
// userservice. The userservice projects away password hashes; a failure here
// is a 502 rather than a silent empty table.
func (s *Server) handleUIAdminUsersJSON(w http.ResponseWriter, r *http.Request) {
c, err := r.Cookie(sessionCookie)
if err != nil {
redirectToLogin(w, r)
return
}
status, body, err := s.callUserserviceAuthed(r.Context(), http.MethodGet, "/users", c.Value)
if err != nil {
s.writeError(w, r, err)
return
}
if status != http.StatusOK {
writeJSON(w, status, map[string]string{"error": "userservice: unexpected response"})
return
}
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write(body)
}
// handleUIAdminSetUserRoleJSON changes a user's role through the userservice
// promote/demote actions.
func (s *Server) handleUIAdminSetUserRoleJSON(w http.ResponseWriter, r *http.Request) {
id, err := uuid.Parse(r.PathValue("id"))
if err != nil {
s.writeError(w, r, domain.ErrInvalidInput)
return
}
var body struct {
Role string `json:"role"`
}
if err := decodeJSON(r, &body); err != nil {
s.writeError(w, r, domain.ErrInvalidInput)
return
}
action := ""
switch body.Role {
case "admin":
action = "promote"
case "user":
action = "demote"
default:
s.writeError(w, r, domain.ErrInvalidInput)
return
}
c, err := r.Cookie(sessionCookie)
if err != nil {
redirectToLogin(w, r)
return
}
status, _, err := s.callUserserviceAuthed(r.Context(), http.MethodPost, "/users/"+id.String()+"/"+action, c.Value)
if err != nil {
s.writeError(w, r, err)
return
}
if status != http.StatusNoContent {
writeJSON(w, status, map[string]string{"error": "userservice: unexpected response"})
return
}
w.WriteHeader(http.StatusNoContent)
}
// handleUIAdminWorkerKeysJSON serves every worker key with its owning user,
// proxied from the userservice.
func (s *Server) handleUIAdminWorkerKeysJSON(w http.ResponseWriter, r *http.Request) {
c, err := r.Cookie(sessionCookie)
if err != nil {
redirectToLogin(w, r)
return
}
status, body, err := s.callUserserviceAuthed(r.Context(), http.MethodGet, "/worker-keys/all", c.Value)
if err != nil {
s.writeError(w, r, err)
return
}
if status != http.StatusOK {
writeJSON(w, status, map[string]string{"error": "userservice: unexpected response"})
return
}
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write(body)
}
// handleUIAdminRevokeKeyJSON revokes any worker key through the userservice
// (whose DELETE endpoint already lets an admin revoke keys of any owner).
func (s *Server) handleUIAdminRevokeKeyJSON(w http.ResponseWriter, r *http.Request) {
id, err := uuid.Parse(r.PathValue("id"))
if err != nil {
s.writeError(w, r, domain.ErrInvalidInput)
return
}
c, err := r.Cookie(sessionCookie)
if err != nil {
redirectToLogin(w, r)
return
}
status, _, err := s.callUserserviceAuthed(r.Context(), http.MethodDelete, "/worker-keys/"+id.String(), c.Value)
if err != nil {
s.writeError(w, r, err)
return
}
if status != http.StatusNoContent {
writeJSON(w, status, map[string]string{"error": "userservice: unexpected response"})
return
}
w.WriteHeader(http.StatusNoContent)
}
// adminOwnerEmails resolves job owner ids to emails through the userservice,
// which is the only place email addresses live. It never blocks the page on
// failure: an empty map leaves the admin jobs table on short ids.
+189 -7
View File
@@ -2,6 +2,7 @@ package usecase
import (
"context"
"log/slog"
"sort"
"strings"
"time"
@@ -9,6 +10,7 @@ import (
"github.com/google/uuid"
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
"github.com/emil28092005/SciMesh/coordinator/internal/workloads"
)
// AdminReadRepository is the bounded read projection behind the coordinator
@@ -49,6 +51,9 @@ type AdminNodeInfo struct {
DBEngine string
PublicURL string
Userservice string // base URL; empty when the UI runs without user auth
// WorkerToken reads the shared worker token for the Settings page. It is a
// func so serve mode can read the token file lazily after provisioning.
WorkerToken func() string
}
type AdminStorageView struct {
@@ -135,18 +140,40 @@ type AdminMetricsView struct {
// Admin answers the coordinator admin console from the bounded read model
// plus process info supplied at startup.
type Admin struct {
read AdminReadRepository
uiRead UIReadRepository
node AdminNodeInfo
ready func(context.Context) error
now func() time.Time
read AdminReadRepository
uiRead UIReadRepository
workers WorkerRepository
settings WorkloadSettingsRepository
catalog *workloads.Catalog
node AdminNodeInfo
ready func(context.Context) error
now func() time.Time
log *slog.Logger
audit func(ctx context.Context, action, detail string)
}
func NewAdmin(read AdminReadRepository, uiRead UIReadRepository, node AdminNodeInfo, ready func(context.Context) error, now func() time.Time) *Admin {
func NewAdmin(read AdminReadRepository, uiRead UIReadRepository, workers WorkerRepository,
settings WorkloadSettingsRepository, catalog *workloads.Catalog, node AdminNodeInfo,
ready func(context.Context) error, now func() time.Time) *Admin {
if now == nil {
now = time.Now
}
return &Admin{read: read, uiRead: uiRead, node: node, ready: ready, now: now}
return &Admin{read: read, uiRead: uiRead, workers: workers, settings: settings, catalog: catalog, node: node, ready: ready, now: now}
}
// WithAuditLog attaches an audit sink for sensitive actions (token reveal).
// Without it the admin usecase stays silent about them.
func (a *Admin) WithAuditLog(log *slog.Logger, audit func(ctx context.Context, action, detail string)) *Admin {
a.log = log
a.audit = audit
return a
}
func (a *Admin) revealToken(ctx context.Context, actor string) string {
if a.node.WorkerToken != nil {
return a.node.WorkerToken()
}
return ""
}
func (a *Admin) System(ctx context.Context) (AdminSystemView, error) {
@@ -339,3 +366,158 @@ func (a *Admin) Metrics(ctx context.Context) (AdminMetricsView, error) {
})
return out, nil
}
// AdminWorkerCard is one row of the admin workers table.
type AdminWorkerCard struct {
ID string `json:"id"`
Name string `json:"name"`
Status string `json:"status"`
Capabilities []string `json:"capabilities"`
Trust string `json:"trust"`
OwnerID string `json:"owner_id,omitempty"`
Owner string `json:"owner"`
Completed int `json:"completed"`
LastHeartbeatAt time.Time `json:"last_heartbeat_at"`
}
type AdminWorkersView struct {
Workers []AdminWorkerCard `json:"workers"`
}
// Workers lists the whole fleet for the admin console. Owner emails are
// resolved through the same map as the jobs table (userservice-backed).
func (a *Admin) Workers(ctx context.Context, ownerEmails map[uuid.UUID]string) (AdminWorkersView, error) {
workers, err := a.uiRead.ListWorkers(ctx, 100)
if err != nil {
return AdminWorkersView{}, err
}
out := AdminWorkersView{Workers: make([]AdminWorkerCard, 0, len(workers))}
for _, w := range workers {
card := AdminWorkerCard{
ID: w.ID.String(),
Name: w.Name,
Status: string(w.Status),
Capabilities: w.Capabilities,
Trust: string(w.TrustLevel),
LastHeartbeatAt: w.LastHeartbeatAt,
Owner: "cluster token",
}
if w.OwnerID != nil {
card.OwnerID = w.OwnerID.String()
card.Owner = "user " + shortID(w.OwnerID.String())
if email, ok := ownerEmails[*w.OwnerID]; ok && email != "" {
card.Owner = email
}
}
out.Workers = append(out.Workers, card)
}
return out, nil
}
// SetTrust reclassifies one worker (trusted/untrusted).
func (a *Admin) SetTrust(ctx context.Context, id uuid.UUID, trusted bool) error {
trust := domain.WorkerUntrusted
if trusted {
trust = domain.WorkerTrusted
}
return a.workers.SetTrust(ctx, id, trust)
}
// AdminWorkloadView is the catalog plus the persisted enable flag.
type AdminWorkloadView struct {
Name string `json:"name"`
Description string `json:"description"`
Reduction string `json:"reduction"`
Parameters int `json:"parameters"`
UploadReady bool `json:"upload_ready"`
Enabled bool `json:"enabled"`
DefaultOn bool `json:"default_on"`
UpdatedAt *time.Time `json:"updated_at,omitempty"`
}
type AdminWorkloadsView struct {
Workloads []AdminWorkloadView `json:"workloads"`
}
// Workloads lists the catalog with persisted enable/disable overrides.
func (a *Admin) Workloads(ctx context.Context) (AdminWorkloadsView, error) {
if a.catalog == nil {
return AdminWorkloadsView{}, domain.ErrInvalidInput
}
items := a.catalog.Items()
overrides, err := a.settings.List(ctx)
if err != nil {
return AdminWorkloadsView{}, err
}
enabled := make(map[string]WorkloadSetting, len(overrides))
for _, s := range overrides {
enabled[s.Workload] = s
}
out := AdminWorkloadsView{Workloads: make([]AdminWorkloadView, 0, len(items))}
for _, item := range items {
params := 0
if properties, ok := item.Parameters["properties"].(map[string]any); ok {
params = len(properties)
}
view := AdminWorkloadView{
Name: item.Name,
Description: item.Description,
Reduction: item.Reduction,
Parameters: params,
UploadReady: item.UploadReady,
Enabled: true,
DefaultOn: true,
}
if s, ok := enabled[item.Name]; ok {
view.Enabled = s.Enabled
view.DefaultOn = false
view.UpdatedAt = &s.UpdatedAt
}
out.Workloads = append(out.Workloads, view)
}
return out, nil
}
// SetWorkloadEnabled flips the persisted enable flag. An unknown workload is
// rejected: the admin console must not invent catalog entries.
func (a *Admin) SetWorkloadEnabled(ctx context.Context, name string, enabled bool) error {
if a.catalog == nil || a.catalog.ByName(name) == nil {
return domain.ErrInvalidInput
}
return a.settings.SetEnabled(ctx, name, enabled, a.now())
}
// AdminSettingsView is the read-only cluster configuration the Settings page
// shows. The token is never included; it is revealed only through
// RevealWorkerToken, which audits.
type AdminSettingsView struct {
PublicURL string `json:"public_url"`
Addr string `json:"addr"`
DataDir string `json:"data_dir"`
DBEngine string `json:"db_engine"`
Binary string `json:"binary"`
}
func (a *Admin) Settings() AdminSettingsView {
return AdminSettingsView{
PublicURL: a.node.PublicURL,
Addr: a.node.Addr,
DataDir: a.node.DataDir,
DBEngine: a.node.DBEngine,
Binary: a.node.Binary,
}
}
// RevealWorkerToken returns the shared worker token for the Settings page and
// records the reveal in the audit log. It must only be called for an admin
// session.
func (a *Admin) RevealWorkerToken(ctx context.Context, actor string) string {
token := a.revealToken(ctx, actor)
if a.audit != nil {
a.audit(ctx, "worker token revealed", "by "+actor)
}
if a.log != nil {
a.log.Warn("admin console revealed the worker token", "actor", actor)
}
return token
}
+115
View File
@@ -9,6 +9,7 @@ import (
"github.com/google/uuid"
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
"github.com/emil28092005/SciMesh/coordinator/internal/workloads"
)
type fakeAdminRead struct {
@@ -73,6 +74,34 @@ type fakeUIRead struct {
workers []domain.Worker
}
type fakeSettings struct {
WorkloadSettingsRepository // embedded: only the methods below are exercised
overrides map[string]bool
}
func (f *fakeSettings) GetEnabled(ctx context.Context, workload string) (bool, error) {
if enabled, ok := f.overrides[workload]; ok {
return enabled, nil
}
return true, nil
}
func (f *fakeSettings) List(ctx context.Context) ([]WorkloadSetting, error) {
out := make([]WorkloadSetting, 0, len(f.overrides))
for name, enabled := range f.overrides {
out = append(out, WorkloadSetting{Workload: name, Enabled: enabled, UpdatedAt: time.Now()})
}
return out, nil
}
func (f *fakeSettings) SetEnabled(ctx context.Context, workload string, enabled bool, now time.Time) error {
if f.overrides == nil {
f.overrides = map[string]bool{}
}
f.overrides[workload] = enabled
return nil
}
func (f *fakeUIRead) ListWorkers(ctx context.Context, limit int) ([]domain.Worker, error) {
return f.workers, nil
}
@@ -81,6 +110,9 @@ func adminFixture() *Admin {
return NewAdmin(
&fakeAdminRead{},
&fakeUIRead{},
nil, // workers repo
&fakeSettings{},
nil, // catalog
AdminNodeInfo{
Version: "1.1.0-alpha.1", StartedAt: time.Unix(1_000_000, 0).UTC(),
Binary: "/usr/local/bin/coordinator", Addr: ":8080", DataDir: "/var/lib/scimesh",
@@ -221,3 +253,86 @@ func TestAdminMetricsBuckets(t *testing.T) {
t.Errorf("rate=%.4f avg=%.2f", v.FailureRate, v.AvgShardSeconds)
}
}
type fakeWorkerRepo struct {
WorkerRepository // embedded: only SetTrust is exercised
}
func (f *fakeWorkerRepo) SetTrust(ctx context.Context, id uuid.UUID, trust domain.WorkerTrust) error {
return nil
}
func TestAdminWorkersAndTrust(t *testing.T) {
owner := uuid.New()
a := adminFixture()
a.uiRead = &fakeUIRead{workers: []domain.Worker{
{ID: uuid.New(), Name: "lab-node-01", Status: domain.WorkerBusy, TrustLevel: domain.WorkerTrusted, Capabilities: []string{"similarity-search"}},
{ID: uuid.New(), Name: "emil-laptop", Status: domain.WorkerOnline, TrustLevel: domain.WorkerUntrusted, OwnerID: &owner},
}}
view, err := a.Workers(context.Background(), map[uuid.UUID]string{owner: "alice@lab.org"})
if err != nil {
t.Fatal(err)
}
if len(view.Workers) != 2 {
t.Fatalf("workers = %d, want 2", len(view.Workers))
}
if view.Workers[0].Trust != "trusted" || view.Workers[1].Trust != "untrusted" {
t.Errorf("trust flags wrong: %+v", view.Workers)
}
if view.Workers[1].Owner != "alice@lab.org" {
t.Errorf("owner = %q, want alice@lab.org", view.Workers[1].Owner)
}
}
func TestAdminSetTrust(t *testing.T) {
called := false
a := adminFixture()
a.workers = &fakeWorkerRepo{}
_ = called
if err := a.SetTrust(context.Background(), uuid.New(), false); err != nil {
t.Fatal(err)
}
}
func TestAdminWorkloadsWithOverrides(t *testing.T) {
catalog, err := workloads.Load()
if err != nil {
t.Fatal(err)
}
a := adminFixture()
a.catalog = catalog
a.settings = &fakeSettings{overrides: map[string]bool{"molwt-filter": false}}
view, err := a.Workloads(context.Background())
if err != nil {
t.Fatal(err)
}
found := false
for _, w := range view.Workloads {
if w.Name == "molwt-filter" {
found = true
if w.Enabled || w.DefaultOn {
t.Errorf("molwt-filter: enabled=%v default_on=%v, want disabled override", w.Enabled, w.DefaultOn)
}
}
if w.Name == "similarity-search" && !w.Enabled {
t.Error("similarity-search must stay enabled (no override)")
}
}
if !found {
t.Error("molwt-filter missing from the catalog view")
}
if err := a.SetWorkloadEnabled(context.Background(), "similarity-search", false); err != nil {
t.Fatal(err)
}
if err := a.SetWorkloadEnabled(context.Background(), "nope", false); err == nil {
t.Error("unknown workload must be rejected")
}
}
func TestAdminRevealToken(t *testing.T) {
a := adminFixture()
a.node.WorkerToken = func() string { return "sm_live_secret" }
if got := a.RevealWorkerToken(context.Background(), "admin:user"); got != "sm_live_secret" {
t.Errorf("token = %q", got)
}
}
+23
View File
@@ -94,6 +94,9 @@ type WorkerRepository interface {
// MarkStaleOffline flips every worker last seen before cutoff to offline and
// reports how many changed.
MarkStaleOffline(ctx context.Context, cutoff time.Time) (int64, error)
// SetTrust reclassifies a worker's trust level (trusted/untrusted). Returns
// ErrNotFound when the id is unknown.
SetTrust(ctx context.Context, id uuid.UUID, trust domain.WorkerTrust) error
}
// ArtifactRepository persists artifact metadata. The bytes live in a BlobStore;
@@ -131,6 +134,26 @@ type Clock interface {
Now() time.Time
}
// WorkloadSetting is one persisted enable/disable override from the admin
// console. A workload with no row in the store is enabled by default.
type WorkloadSetting struct {
Workload string `json:"workload"`
Enabled bool `json:"enabled"`
UpdatedAt time.Time `json:"updated_at"`
}
// WorkloadSettingsRepository persists the admin enable/disable overrides on
// top of the embedded workload catalog.
type WorkloadSettingsRepository interface {
// GetEnabled reports whether the workload is enabled. True when the
// workload has no override row (catalog default).
GetEnabled(ctx context.Context, workload string) (bool, error)
// List returns every override row, newest update first.
List(ctx context.Context) ([]WorkloadSetting, error)
// SetEnabled upserts the override.
SetEnabled(ctx context.Context, workload string, enabled bool, now time.Time) error
}
// ErrNotImplemented marks scaffold code with no body yet. Unlike the errors in
// domain, it describes the state of this codebase, not a business rule.
var ErrNotImplemented = errors.New("not implemented")
+12 -2
View File
@@ -24,17 +24,27 @@ type SubmitDataset struct {
clk Clock
maxAttempts int
catalog *workloads.Catalog
settings WorkloadSettingsRepository
}
func NewSubmitDataset(blobs BlobStore, artifacts ArtifactRepository, jobs JobRepository,
tasks TaskRepository, tx TxManager, clk Clock, maxAttempts int, catalog *workloads.Catalog) *SubmitDataset {
return &SubmitDataset{blobs: blobs, artifacts: artifacts, jobs: jobs, tasks: tasks, tx: tx, clk: clk, maxAttempts: maxAttempts, catalog: catalog}
tasks TaskRepository, tx TxManager, clk Clock, maxAttempts int, catalog *workloads.Catalog, settings WorkloadSettingsRepository) *SubmitDataset {
return &SubmitDataset{blobs: blobs, artifacts: artifacts, jobs: jobs, tasks: tasks, tx: tx, clk: clk, maxAttempts: maxAttempts, catalog: catalog, settings: settings}
}
func (uc *SubmitDataset) Execute(ctx context.Context, in SubmitDatasetInput) (SubmitDatasetResult, error) {
if err := validateUploadedWorkload(uc.catalog, in.Workload, in.Parameters); err != nil {
return SubmitDatasetResult{}, err
}
if uc.settings != nil {
enabled, err := uc.settings.GetEnabled(ctx, in.Workload)
if err != nil {
return SubmitDatasetResult{}, err
}
if !enabled {
return SubmitDatasetResult{}, domain.ErrWorkloadDisabled
}
}
if uc.maxAttempts < 1 {
return SubmitDatasetResult{}, domain.ErrInvalidInput
}
+46 -1
View File
@@ -42,6 +42,7 @@ type harness struct {
blobs *memstore.BlobStore
clk *memstore.Clock
taskResults *memstore.TaskResultRepo
settings *memSettings
createJob *usecase.CreateJob
submit *usecase.SubmitDataset
@@ -70,10 +71,11 @@ func newHarness() *harness {
blobs: memstore.NewBlobStore(),
clk: memstore.NewClock(time.Date(2026, 7, 21, 12, 0, 0, 0, time.UTC)),
taskResults: memstore.NewTaskResultRepo(),
settings: newMemSettings(),
}
tx := memstore.Tx{}
h.createJob = usecase.NewCreateJob(h.jobs, h.tasks, tx, h.clk)
h.submit = usecase.NewSubmitDataset(h.blobs, h.arts, h.jobs, h.tasks, tx, h.clk, 3, testCatalog())
h.submit = usecase.NewSubmitDataset(h.blobs, h.arts, h.jobs, h.tasks, tx, h.clk, 3, testCatalog(), h.settings)
h.claim = usecase.NewClaimTask(h.tasks, h.jobs, h.work, tx, h.clk, lease, testCatalog())
h.renew = usecase.NewRenewLease(h.tasks, h.work, tx, h.clk, lease)
h.complete = usecase.NewCompleteTask(h.tasks, h.jobs, h.arts, h.work, h.taskResults, tx, h.clk, 2, testCatalog())
@@ -935,3 +937,46 @@ func TestFinalLeaseExpiryPersistsFailedJobAndCannotBeCancelled(t *testing.T) {
t.Errorf("cancel terminal lease failure = %v, want ErrJobNotCancellable", err)
}
}
// memSettings is an in-memory WorkloadSettingsRepository for tests.
type memSettings struct {
overrides map[string]bool
}
func newMemSettings() *memSettings { return &memSettings{overrides: map[string]bool{}} }
func (m *memSettings) GetEnabled(ctx context.Context, workload string) (bool, error) {
if enabled, ok := m.overrides[workload]; ok {
return enabled, nil
}
return true, nil
}
func (m *memSettings) List(ctx context.Context) ([]usecase.WorkloadSetting, error) { return nil, nil }
func (m *memSettings) SetEnabled(ctx context.Context, workload string, enabled bool, now time.Time) error {
m.overrides[workload] = enabled
return nil
}
func TestSubmitDatasetRejectsDisabledWorkload(t *testing.T) {
h := newHarness()
h.settings.SetEnabled(ctx, "molwt-filter", false, h.clk.Now())
_, err := h.submit.Execute(ctx, usecase.SubmitDatasetInput{
Workload: "molwt-filter", Parameters: map[string]any{"min_molwt": 100, "max_molwt": 600},
RowsPerShard: 2, Filename: "m.tsv", ContentType: "text/tab-separated-values",
Body: strings.NewReader("smiles\nCC\n"),
})
if !errors.Is(err, domain.ErrWorkloadDisabled) {
t.Fatalf("err = %v, want ErrWorkloadDisabled", err)
}
// Re-enabling accepts the same submit.
h.settings.SetEnabled(ctx, "molwt-filter", true, h.clk.Now())
if _, err := h.submit.Execute(ctx, usecase.SubmitDatasetInput{
Workload: "molwt-filter", Parameters: map[string]any{"min_molwt": 100, "max_molwt": 600},
RowsPerShard: 2, Filename: "m.tsv", ContentType: "text/tab-separated-values",
Body: strings.NewReader("chembl_id\tcanonical_smiles\nA\tCC\n"),
}); err != nil {
t.Fatalf("submit after re-enable: %v", err)
}
}
+12 -9
View File
@@ -51,15 +51,18 @@ func Serve(ctx context.Context, cfg Config) (string, func() error, error) {
issuer := auth.NewIssuer(cfg.JWTSecret, 24*time.Hour, clock.Now)
uc := usershttp.UseCases{
Register: usecase.NewRegister(users, hasher, clock),
Login: usecase.NewLogin(users, hasher, issuer),
SetVerified: usecase.NewSetVerified(users),
SetRole: usecase.NewSetRole(users),
CreateWorkerKey: usecase.NewCreateWorkerKey(workerKeys, clock),
ListWorkerKeys: usecase.NewListWorkerKeys(workerKeys),
RevokeWorkerKey: usecase.NewRevokeWorkerKey(workerKeys),
ExchangeWorkerKey: usecase.NewExchangeWorkerKey(workerKeys, users, issuer, 24*time.Hour),
Users: users,
Register: usecase.NewRegister(users, hasher, clock),
Login: usecase.NewLogin(users, hasher, issuer),
SetVerified: usecase.NewSetVerified(users),
SetRole: usecase.NewSetRole(users),
CreateWorkerKey: usecase.NewCreateWorkerKey(workerKeys, clock),
ListWorkerKeys: usecase.NewListWorkerKeys(workerKeys),
ListWorkerKeysAll: usecase.NewListWorkerKeysAll(workerKeys),
RevokeWorkerKey: usecase.NewRevokeWorkerKey(workerKeys),
RevokeWorkerKeyAdmin: usecase.NewRevokeWorkerKeyAdmin(workerKeys),
ExchangeWorkerKey: usecase.NewExchangeWorkerKey(workerKeys, users, issuer, 24*time.Hour),
ListUsers: usecase.NewListUsers(users),
Users: users,
}
if cfg.AdminEmail != "" && cfg.AdminPassword != "" {
@@ -4,6 +4,7 @@ package memstore
import (
"context"
"sort"
"sync"
"time"
@@ -84,7 +85,105 @@ func (r *UserRepo) SetRole(_ context.Context, id uuid.UUID, role domain.Role) er
return nil
}
// ListUsers returns every account, oldest first. It copies, so callers cannot
// corrupt the store through the returned slice.
func (r *UserRepo) ListUsers(_ context.Context) ([]*domain.User, error) {
r.mu.Lock()
defer r.mu.Unlock()
users := make([]*domain.User, 0, len(r.byID))
for _, u := range r.byID {
copy := u
users = append(users, &copy)
}
sort.Slice(users, func(i, j int) bool { return users[i].CreatedAt.Before(users[j].CreatedAt) })
return users, nil
}
// Clock is a fixed usecase.Clock for deterministic tests.
type Clock struct{ T time.Time }
func (c Clock) Now() time.Time { return c.T }
// WorkerKeyRepo is an in-memory usecase.WorkerKeyRepository.
type WorkerKeyRepo struct {
mu sync.Mutex
keys map[uuid.UUID]*domain.WorkerKey
}
func NewWorkerKeyRepo() *WorkerKeyRepo {
return &WorkerKeyRepo{keys: map[uuid.UUID]*domain.WorkerKey{}}
}
func (r *WorkerKeyRepo) Insert(_ context.Context, k *domain.WorkerKey) error {
r.mu.Lock()
defer r.mu.Unlock()
r.keys[k.ID] = k
return nil
}
func (r *WorkerKeyRepo) ListByUser(_ context.Context, userID uuid.UUID) ([]*domain.WorkerKey, error) {
r.mu.Lock()
defer r.mu.Unlock()
var out []*domain.WorkerKey
for _, k := range r.keys {
if k.UserID == userID && !k.Revoked() {
out = append(out, k)
}
}
return out, nil
}
func (r *WorkerKeyRepo) ListAll(_ context.Context) ([]*domain.WorkerKey, error) {
r.mu.Lock()
defer r.mu.Unlock()
out := make([]*domain.WorkerKey, 0, len(r.keys))
for _, k := range r.keys {
out = append(out, k)
}
return out, nil
}
func (r *WorkerKeyRepo) GetActiveByHash(_ context.Context, tokenHash string) (*domain.WorkerKey, error) {
r.mu.Lock()
defer r.mu.Unlock()
for _, k := range r.keys {
if k.TokenHash == tokenHash && !k.Revoked() {
return k, nil
}
}
return nil, usecase.ErrWorkerKeyNotFound
}
func (r *WorkerKeyRepo) Revoke(_ context.Context, id, userID uuid.UUID) error {
r.mu.Lock()
defer r.mu.Unlock()
k, ok := r.keys[id]
if !ok || k.UserID != userID || k.Revoked() {
return usecase.ErrWorkerKeyNotFound
}
now := time.Now()
k.RevokedAt = &now
return nil
}
func (r *WorkerKeyRepo) RevokeAny(_ context.Context, id uuid.UUID) error {
r.mu.Lock()
defer r.mu.Unlock()
k, ok := r.keys[id]
if !ok || k.Revoked() {
return usecase.ErrWorkerKeyNotFound
}
now := time.Now()
k.RevokedAt = &now
return nil
}
func (r *WorkerKeyRepo) TouchLastUsed(_ context.Context, id uuid.UUID) error {
r.mu.Lock()
defer r.mu.Unlock()
if k, ok := r.keys[id]; ok {
now := time.Now()
k.LastUsedAt = &now
}
return nil
}
@@ -93,6 +93,24 @@ func (r *UserRepo) SetRole(ctx context.Context, id uuid.UUID, role domain.Role)
return rowsAffectedOrNotFound(res, usecase.ErrUserNotFound)
}
// ListUsers returns every account, oldest first.
func (r *UserRepo) ListUsers(ctx context.Context) ([]*domain.User, error) {
rows, err := r.db.QueryContext(ctx, "SELECT "+userColumns+" FROM users ORDER BY created_at ASC, id ASC")
if err != nil {
return nil, err
}
defer func() { _ = rows.Close() }()
var users []*domain.User
for rows.Next() {
user, err := scanUser(rows)
if err != nil {
return nil, err
}
users = append(users, user)
}
return users, rows.Err()
}
const workerKeyColumns = `id, user_id, name, token_hash, prefix, created_at, last_used_at, revoked_at`
func scanWorkerKey(row interface{ Scan(dest ...any) error }) (*domain.WorkerKey, error) {
@@ -154,6 +172,25 @@ func (r *WorkerKeyRepo) ListByUser(ctx context.Context, userID uuid.UUID) ([]*do
return keys, rows.Err()
}
// ListAll returns every key, revoked included, newest first. Admin-only.
func (r *WorkerKeyRepo) ListAll(ctx context.Context) ([]*domain.WorkerKey, error) {
rows, err := r.db.QueryContext(ctx,
"SELECT "+workerKeyColumns+" FROM worker_keys ORDER BY created_at DESC, id DESC")
if err != nil {
return nil, err
}
defer func() { _ = rows.Close() }()
var keys []*domain.WorkerKey
for rows.Next() {
key, err := scanWorkerKey(rows)
if err != nil {
return nil, err
}
keys = append(keys, key)
}
return keys, rows.Err()
}
func (r *WorkerKeyRepo) GetActiveByHash(ctx context.Context, tokenHash string) (*domain.WorkerKey, error) {
row := r.db.QueryRowContext(ctx,
"SELECT "+workerKeyColumns+" FROM worker_keys WHERE token_hash = ? AND revoked_at IS NULL",
@@ -172,6 +209,17 @@ func (r *WorkerKeyRepo) Revoke(ctx context.Context, id, userID uuid.UUID) error
return rowsAffectedOrNotFound(res, usecase.ErrWorkerKeyNotFound)
}
// RevokeAny retires a key by id regardless of its owner.
func (r *WorkerKeyRepo) RevokeAny(ctx context.Context, id uuid.UUID) error {
res, err := r.db.ExecContext(ctx,
"UPDATE worker_keys SET revoked_at = ? WHERE id = ? AND revoked_at IS NULL",
time.Now().UnixNano(), id.String())
if err != nil {
return err
}
return rowsAffectedOrNotFound(res, usecase.ErrWorkerKeyNotFound)
}
func (r *WorkerKeyRepo) TouchLastUsed(ctx context.Context, id uuid.UUID) error {
_, err := r.db.ExecContext(ctx,
"UPDATE worker_keys SET last_used_at = ? WHERE id = ?",
@@ -79,6 +79,27 @@ type workerKeysResponse struct {
WorkerKeys []workerKeyResponse `json:"worker_keys"`
}
// adminWorkerKeyResponse extends the public view with the owning user and the
// revocation state, both needed by the coordinator admin console.
type adminWorkerKeyResponse struct {
workerKeyResponse
UserID string `json:"user_id"`
RevokedAt string `json:"revoked_at,omitempty"`
}
func toAdminWorkerKeyResponse(k *domain.WorkerKey) adminWorkerKeyResponse {
resp := adminWorkerKeyResponse{workerKeyResponse: toWorkerKeyResponse(k), UserID: k.UserID.String()}
if k.RevokedAt != nil {
resp.RevokedAt = k.RevokedAt.UTC().Format(time.RFC3339)
}
return resp
}
// usersResponse is the admin list of accounts, password hashes excluded.
type usersResponse struct {
Users []userResponse `json:"users"`
}
func toWorkerKeyResponse(k *domain.WorkerKey) workerKeyResponse {
resp := workerKeyResponse{
ID: k.ID.String(),
@@ -14,16 +14,19 @@ import (
// Handlers holds the use cases each endpoint drives.
type Handlers struct {
register *usecase.Register
login *usecase.Login
setVerified *usecase.SetVerified
setRole *usecase.SetRole
createWorkerKey *usecase.CreateWorkerKey
listWorkerKeys *usecase.ListWorkerKeys
revokeWorkerKey *usecase.RevokeWorkerKey
exchangeWorkerKey *usecase.ExchangeWorkerKey
users usecase.UserRepository
log *slog.Logger
register *usecase.Register
login *usecase.Login
setVerified *usecase.SetVerified
setRole *usecase.SetRole
createWorkerKey *usecase.CreateWorkerKey
listWorkerKeys *usecase.ListWorkerKeys
listWorkerKeysAll *usecase.ListWorkerKeysAll
revokeWorkerKey *usecase.RevokeWorkerKey
revokeWorkerKeyAdmin *usecase.RevokeWorkerKeyAdmin
exchangeWorkerKey *usecase.ExchangeWorkerKey
listUsers *usecase.ListUsers
users usecase.UserRepository
log *slog.Logger
}
// handleHealth is an unauthenticated liveness probe for the container and load
@@ -173,15 +176,42 @@ func (h *Handlers) handleListWorkerKeys(w http.ResponseWriter, r *http.Request)
writeJSON(w, http.StatusOK, workerKeysResponse{WorkerKeys: out})
}
// handleListWorkerKeysAll returns every key in the service — revoked included,
// with the owning user id — for the coordinator admin console. Admin-only.
func (h *Handlers) handleListWorkerKeysAll(w http.ResponseWriter, r *http.Request) {
keys, err := h.listWorkerKeysAll.Execute(r.Context())
if err != nil {
writeError(w, r, h.log, err)
return
}
out := make([]adminWorkerKeyResponse, 0, len(keys))
for _, k := range keys {
out = append(out, toAdminWorkerKeyResponse(k))
}
writeJSON(w, http.StatusOK, struct {
WorkerKeys []adminWorkerKeyResponse `json:"worker_keys"`
}{WorkerKeys: out})
}
// handleListUsers returns every account for the coordinator admin console.
// Password hashes never leave the service: only the public projection is sent.
func (h *Handlers) handleListUsers(w http.ResponseWriter, r *http.Request) {
users, err := h.listUsers.Execute(r.Context())
if err != nil {
writeError(w, r, h.log, err)
return
}
out := make([]userResponse, 0, len(users))
for _, u := range users {
out = append(out, toUserResponse(u))
}
writeJSON(w, http.StatusOK, usersResponse{Users: out})
}
// handleRevokeWorkerKey retires one of the caller's keys. The repository scopes
// the delete to the owner, so a mismatched id is a clean 404, not another user's
// key.
func (h *Handlers) handleRevokeWorkerKey(w http.ResponseWriter, r *http.Request) {
userID, ok := userIDFrom(r.Context())
if !ok {
unauthorized(w, r)
return
}
keyID, err := uuid.Parse(r.PathValue("id"))
if err != nil {
writeJSON(w, http.StatusBadRequest, errorResponse{
@@ -190,6 +220,20 @@ func (h *Handlers) handleRevokeWorkerKey(w http.ResponseWriter, r *http.Request)
})
return
}
// An admin may revoke any key; a plain user only their own.
if role, ok := r.Context().Value(roleKey).(domain.Role); ok && role == domain.RoleAdmin {
if err := h.revokeWorkerKeyAdmin.Execute(r.Context(), keyID); err != nil {
writeError(w, r, h.log, err)
return
}
w.WriteHeader(http.StatusNoContent)
return
}
userID, ok := userIDFrom(r.Context())
if !ok {
unauthorized(w, r)
return
}
if err := h.revokeWorkerKey.Execute(r.Context(), userID, keyID); err != nil {
writeError(w, r, h.log, err)
return
@@ -14,31 +14,37 @@ import (
// UseCases bundles the application services the handlers drive.
type UseCases struct {
Register *usecase.Register
Login *usecase.Login
SetVerified *usecase.SetVerified
SetRole *usecase.SetRole
CreateWorkerKey *usecase.CreateWorkerKey
ListWorkerKeys *usecase.ListWorkerKeys
RevokeWorkerKey *usecase.RevokeWorkerKey
ExchangeWorkerKey *usecase.ExchangeWorkerKey
Users usecase.UserRepository
Register *usecase.Register
Login *usecase.Login
SetVerified *usecase.SetVerified
SetRole *usecase.SetRole
CreateWorkerKey *usecase.CreateWorkerKey
ListWorkerKeys *usecase.ListWorkerKeys
ListWorkerKeysAll *usecase.ListWorkerKeysAll
RevokeWorkerKey *usecase.RevokeWorkerKey
RevokeWorkerKeyAdmin *usecase.RevokeWorkerKeyAdmin
ExchangeWorkerKey *usecase.ExchangeWorkerKey
ListUsers *usecase.ListUsers
Users usecase.UserRepository
}
// NewServer wires the routes and the middleware stack and returns the handler.
// The issuer verifies tokens for the JWT-protected routes.
func NewServer(log *slog.Logger, uc UseCases, issuer auth.Issuer) http.Handler {
h := &Handlers{
register: uc.Register,
login: uc.Login,
setVerified: uc.SetVerified,
setRole: uc.SetRole,
createWorkerKey: uc.CreateWorkerKey,
listWorkerKeys: uc.ListWorkerKeys,
revokeWorkerKey: uc.RevokeWorkerKey,
exchangeWorkerKey: uc.ExchangeWorkerKey,
users: uc.Users,
log: log,
register: uc.Register,
login: uc.Login,
setVerified: uc.SetVerified,
setRole: uc.SetRole,
createWorkerKey: uc.CreateWorkerKey,
listWorkerKeys: uc.ListWorkerKeys,
listWorkerKeysAll: uc.ListWorkerKeysAll,
revokeWorkerKey: uc.RevokeWorkerKey,
revokeWorkerKeyAdmin: uc.RevokeWorkerKeyAdmin,
exchangeWorkerKey: uc.ExchangeWorkerKey,
listUsers: uc.ListUsers,
users: uc.Users,
log: log,
}
mux := http.NewServeMux()
@@ -68,6 +74,12 @@ func NewServer(log *slog.Logger, uc UseCases, issuer auth.Issuer) http.Handler {
mux.Handle("POST /users/{id}/demote",
chain(h.handleSetRole(domain.RoleUser), withJWT(issuer), withAdmin))
// Admin console: lists of every account and every worker key, and the key
// revoke path the admin console calls (the same DELETE endpoint already
// lets an admin revoke any key).
mux.Handle("GET /users", chain(http.HandlerFunc(h.handleListUsers), withJWT(issuer), withAdmin))
mux.Handle("GET /worker-keys/all", chain(http.HandlerFunc(h.handleListWorkerKeysAll), withJWT(issuer), withAdmin))
// Outermost first: every request gets an ID and an access-log line.
return chain(mux, withRequestID, withAccessLog(log))
}
@@ -25,17 +25,25 @@ const secret = "server-test-secret-32-bytes-long!!!!"
func newTestServer() http.Handler {
users := memstore.NewUserRepo()
keys := memstore.NewWorkerKeyRepo()
hasher := auth.NewHasher(4)
clk := memstore.Clock{T: time.Date(2026, 7, 26, 0, 0, 0, 0, time.UTC)}
// Real clock for the issuer so tokens are valid at verification time.
issuer := auth.NewIssuer(secret, time.Hour, nil)
uc := apihttp.UseCases{
Register: usecase.NewRegister(users, hasher, clk),
Login: usecase.NewLogin(users, hasher, issuer),
SetVerified: usecase.NewSetVerified(users),
SetRole: usecase.NewSetRole(users),
Users: users,
Register: usecase.NewRegister(users, hasher, clk),
Login: usecase.NewLogin(users, hasher, issuer),
SetVerified: usecase.NewSetVerified(users),
SetRole: usecase.NewSetRole(users),
CreateWorkerKey: usecase.NewCreateWorkerKey(keys, clk),
ListWorkerKeys: usecase.NewListWorkerKeys(keys),
ListWorkerKeysAll: usecase.NewListWorkerKeysAll(keys),
RevokeWorkerKey: usecase.NewRevokeWorkerKey(keys),
RevokeWorkerKeyAdmin: usecase.NewRevokeWorkerKeyAdmin(keys),
ExchangeWorkerKey: usecase.NewExchangeWorkerKey(keys, users, issuer, time.Hour),
ListUsers: usecase.NewListUsers(users),
Users: users,
}
log := slog.New(slog.NewTextHandler(io.Discard, nil))
return apihttp.NewServer(log, uc, issuer)
@@ -216,10 +224,16 @@ func TestMeInternalError(t *testing.T) {
}
// mintToken issues a token with the package secret for a synthetic caller of the
// given role — enough to drive the admin-gated endpoints.
// given role — enough to drive the admin-gated endpoints. userID defaults to a
// fresh random id; pass one to act as an existing account.
func mintToken(t *testing.T, role domain.Role) string {
t.Helper()
token, err := auth.NewIssuer(secret, time.Hour, nil).Issue(&domain.User{ID: uuid.New(), Role: role})
return mintTokenFor(t, role, uuid.New())
}
func mintTokenFor(t *testing.T, role domain.Role, userID uuid.UUID) string {
t.Helper()
token, err := auth.NewIssuer(secret, time.Hour, nil).Issue(&domain.User{ID: userID, Role: role})
if err != nil {
t.Fatal(err)
}
@@ -370,3 +384,98 @@ func TestUnverifyRevokes(t *testing.T) {
t.Error("verified should be false after unverify")
}
}
func TestAdminListsUsersAndKeys(t *testing.T) {
h := newTestServer()
userID, _ := uuid.Parse(registerUser(t, h, "listed@example.com"))
userToken := mintTokenFor(t, domain.RoleUser, userID)
// Mint a worker key as the plain user.
keyRec := do(t, h, http.MethodPost, "/worker-keys", userToken, map[string]string{"name": "lab-node"})
if keyRec.Code != http.StatusCreated {
t.Fatalf("create key: got %d, body %s", keyRec.Code, keyRec.Body)
}
var created struct {
ID string `json:"id"`
}
_ = json.Unmarshal(keyRec.Body.Bytes(), &created)
// Admin lists users: emails present, password hashes absent.
rec := do(t, h, http.MethodGet, "/users", mintToken(t, domain.RoleAdmin), nil)
if rec.Code != http.StatusOK {
t.Fatalf("list users: got %d, body %s", rec.Code, rec.Body)
}
var users struct {
Users []struct {
Email string `json:"email"`
Role string `json:"role"`
PasswordHash string `json:"password_hash"`
} `json:"users"`
}
if err := json.Unmarshal(rec.Body.Bytes(), &users); err != nil {
t.Fatal(err)
}
found := false
for _, u := range users.Users {
if u.PasswordHash != "" {
t.Error("password hash leaked through the admin users list")
}
if u.Email == "listed@example.com" {
found = true
}
}
if !found {
t.Error("listed user missing from the admin list")
}
// Admin lists all keys: the owner is attached, no secret.
rec = do(t, h, http.MethodGet, "/worker-keys/all", mintToken(t, domain.RoleAdmin), nil)
if rec.Code != http.StatusOK {
t.Fatalf("list all keys: got %d", rec.Code)
}
var keys struct {
WorkerKeys []struct {
ID string `json:"id"`
UserID string `json:"user_id"`
Name string `json:"name"`
} `json:"worker_keys"`
}
if err := json.Unmarshal(rec.Body.Bytes(), &keys); err != nil {
t.Fatal(err)
}
if len(keys.WorkerKeys) != 1 || keys.WorkerKeys[0].UserID != userID.String() {
t.Errorf("all keys = %+v, want the one key owned by %s", keys.WorkerKeys, userID)
}
// Plain users cannot see either list.
for _, path := range []string{"/users", "/worker-keys/all"} {
if rec := do(t, h, http.MethodGet, path, mintToken(t, domain.RoleUser), nil); rec.Code != http.StatusForbidden {
t.Errorf("%s as user: got %d, want 403", path, rec.Code)
}
}
// An admin revokes a key that belongs to another user; the plain owner of
// that key could not (it would be a 404, scoped to their own keys).
// The owner of the key revokes it themselves: 204.
if rec := do(t, h, http.MethodDelete, "/worker-keys/"+created.ID, userToken, nil); rec.Code != http.StatusNoContent {
t.Errorf("user revoke own key: got %d, want 204", rec.Code)
}
// Another plain user cannot revoke it: scoped to their own keys, so a
// mismatch reads as 404.
otherID, _ := uuid.Parse(registerUser(t, h, "other@example.com"))
if rec := do(t, h, http.MethodDelete, "/worker-keys/"+created.ID, mintTokenFor(t, domain.RoleUser, otherID), nil); rec.Code != http.StatusNotFound {
t.Errorf("other user revoke: got %d, want 404", rec.Code)
}
// An admin revokes a key that belongs to someone else: 204.
keyRec = do(t, h, http.MethodPost, "/worker-keys", userToken, map[string]string{"name": "lab-node-2"})
var second struct {
ID string `json:"id"`
}
_ = json.Unmarshal(keyRec.Body.Bytes(), &second)
if rec := do(t, h, http.MethodDelete, "/worker-keys/"+second.ID, mintToken(t, domain.RoleAdmin), nil); rec.Code != http.StatusNoContent {
t.Errorf("admin revoke other's key: got %d, want 204", rec.Code)
}
// Admin cannot revoke an unknown key.
if rec := do(t, h, http.MethodDelete, "/worker-keys/"+uuid.NewString(), mintToken(t, domain.RoleAdmin), nil); rec.Code != http.StatusNotFound {
t.Errorf("admin revoke unknown key: got %d, want 404", rec.Code)
}
}
@@ -35,6 +35,7 @@ func (s stubRepo) SetVerified(context.Context, uuid.UUID, bool) error {
func (s stubRepo) SetRole(context.Context, uuid.UUID, domain.Role) error {
return usecase.ErrUserNotFound
}
func (s stubRepo) ListUsers(context.Context) ([]*domain.User, error) { return nil, nil }
type stubHasher struct {
hashErr error
@@ -28,6 +28,10 @@ type UserRepository interface {
// SetRole changes a user's role, returning ErrUserNotFound if no such user
// exists.
SetRole(ctx context.Context, id uuid.UUID, role domain.Role) error
// ListUsers returns every account, oldest first. Admin-only: used by the
// coordinator admin console; the response must never carry password hashes
// (the caller projects the entity).
ListUsers(ctx context.Context) ([]*domain.User, error)
}
// WorkerKeyRepository persists and looks up the long-lived worker keys a user
@@ -38,12 +42,18 @@ type WorkerKeyRepository interface {
Insert(ctx context.Context, k *domain.WorkerKey) error
// ListByUser returns a user's live (non-revoked) keys, newest first.
ListByUser(ctx context.Context, userID uuid.UUID) ([]*domain.WorkerKey, error)
// ListAll returns every key (revoked included), newest first. Admin-only:
// backs the coordinator admin console's key table.
ListAll(ctx context.Context) ([]*domain.WorkerKey, error)
// GetActiveByHash returns the non-revoked key with the given hash, or
// ErrWorkerKeyNotFound.
GetActiveByHash(ctx context.Context, tokenHash string) (*domain.WorkerKey, error)
// Revoke retires a key the user owns, returning ErrWorkerKeyNotFound when no
// live key with that id belongs to the user.
Revoke(ctx context.Context, id, userID uuid.UUID) error
// RevokeAny retires a key by id regardless of its owner. Admin-only; the
// coordinator admin console uses it to cut a key immediately.
RevokeAny(ctx context.Context, id uuid.UUID) error
// TouchLastUsed records a successful exchange. Best-effort: a failure here
// must not fail the exchange itself.
TouchLastUsed(ctx context.Context, id uuid.UUID) error
@@ -0,0 +1,21 @@
package usecase
import (
"context"
"github.com/emil28092005/SciMesh/coordinator/internal/userservice/domain"
)
// ListUsers returns every account for the coordinator admin console. The
// handler must project the entities so password hashes never leave the service.
type ListUsers struct {
users UserRepository
}
func NewListUsers(users UserRepository) *ListUsers {
return &ListUsers{users: users}
}
func (uc *ListUsers) Execute(ctx context.Context) ([]*domain.User, error) {
return uc.users.ListUsers(ctx)
}
@@ -47,6 +47,34 @@ func (uc *ListWorkerKeys) Execute(ctx context.Context, userID uuid.UUID) ([]*dom
return uc.keys.ListByUser(ctx, userID)
}
// ListWorkerKeysAll returns every key in the service, revoked included, for
// the coordinator admin console. Admin-only.
type ListWorkerKeysAll struct {
keys WorkerKeyRepository
}
func NewListWorkerKeysAll(keys WorkerKeyRepository) *ListWorkerKeysAll {
return &ListWorkerKeysAll{keys: keys}
}
func (uc *ListWorkerKeysAll) Execute(ctx context.Context) ([]*domain.WorkerKey, error) {
return uc.keys.ListAll(ctx)
}
// RevokeWorkerKeyAdmin retires any key, regardless of owner. Admin-only; used
// by the coordinator admin console when a key must be cut immediately.
type RevokeWorkerKeyAdmin struct {
keys WorkerKeyRepository
}
func NewRevokeWorkerKeyAdmin(keys WorkerKeyRepository) *RevokeWorkerKeyAdmin {
return &RevokeWorkerKeyAdmin{keys: keys}
}
func (uc *RevokeWorkerKeyAdmin) Execute(ctx context.Context, id uuid.UUID) error {
return uc.keys.RevokeAny(ctx, id)
}
// RevokeWorkerKey retires one of the caller's keys.
type RevokeWorkerKey struct {
keys WorkerKeyRepository
@@ -64,6 +64,24 @@ func (r *fakeKeyRepo) TouchLastUsed(_ context.Context, id uuid.UUID) error {
return nil
}
func (r *fakeKeyRepo) ListAll(_ context.Context) ([]*domain.WorkerKey, error) {
out := make([]*domain.WorkerKey, 0, len(r.byID))
for _, k := range r.byID {
out = append(out, k)
}
return out, nil
}
func (r *fakeKeyRepo) RevokeAny(_ context.Context, id uuid.UUID) error {
k, ok := r.byID[id]
if !ok || k.Revoked() {
return usecase.ErrWorkerKeyNotFound
}
now := time.Now()
k.RevokedAt = &now
return nil
}
func newKeyFixtures(t *testing.T) (*usecase.CreateWorkerKey, *usecase.ExchangeWorkerKey, *usecase.RevokeWorkerKey, *usecase.ListWorkerKeys, *fakeKeyRepo, *domain.User) {
t.Helper()
users := memstore.NewUserRepo()
@@ -176,6 +176,12 @@ func (c *Catalog) Enabled() []*Workload {
return result
}
// Items returns every workload in the catalog, sorted by name (the same
// ordering as Enabled). The caller must not mutate the entries.
func (c *Catalog) Items() []*Workload {
return c.workloads
}
// ByName returns the workload with the given name, or nil.
func (c *Catalog) ByName(name string) *Workload {
for _, workload := range c.workloads {