From e923627ce0910a25e1cfe441c453fb603aec029f Mon Sep 17 00:00:00 2001 From: Emil Date: Mon, 3 Aug 2026 00:11:47 +0300 Subject: [PATCH] Add admin console access and platform pages (trust, users, keys, workloads, settings) --- coordinator/cmd/coordinator/main.go | 30 ++- coordinator/internal/domain/errors.go | 1 + coordinator/internal/memstore/memstore.go | 11 + .../internal/storage/postgres/migrate_test.go | 2 + .../0014_workload_settings.down.sql | 5 + .../migrations/0014_workload_settings.up.sql | 11 + .../internal/storage/postgres/worker_repo.go | 18 ++ .../postgres/workload_settings_repo.go | 67 ++++++ .../internal/storage/sqlite/admin_m2_test.go | 89 ++++++++ .../migrations/0002_workload_settings.sql | 8 + .../internal/storage/sqlite/sqlite_test.go | 4 +- .../internal/storage/sqlite/worker_repo.go | 17 ++ .../storage/sqlite/workload_settings_repo.go | 75 +++++++ coordinator/internal/transport/http/errors.go | 2 +- coordinator/internal/transport/http/server.go | 10 + .../internal/transport/http/server_test.go | 22 +- .../transport/http/templates/admin.html | 167 ++++++++++++++- .../internal/transport/http/ui_admin.go | 194 +++++++++++++++++ coordinator/internal/usecase/admin.go | 196 +++++++++++++++++- coordinator/internal/usecase/admin_test.go | 115 ++++++++++ coordinator/internal/usecase/ports.go | 23 ++ coordinator/internal/usecase/upload.go | 14 +- coordinator/internal/usecase/usecase_test.go | 47 ++++- coordinator/internal/userservice/embedded.go | 21 +- .../internal/userservice/memstore/memstore.go | 99 +++++++++ .../userservice/storage/sqlite/sqlite.go | 48 +++++ .../userservice/transport/http/dto.go | 21 ++ .../userservice/transport/http/handlers.go | 74 +++++-- .../userservice/transport/http/server.go | 50 +++-- .../userservice/transport/http/server_test.go | 123 ++++++++++- .../userservice/usecase/errorpaths_test.go | 1 + .../internal/userservice/usecase/ports.go | 10 + .../internal/userservice/usecase/users.go | 21 ++ .../internal/userservice/usecase/workerkey.go | 28 +++ .../userservice/usecase/workerkey_test.go | 18 ++ coordinator/internal/workloads/catalog.go | 6 + 36 files changed, 1562 insertions(+), 86 deletions(-) create mode 100644 coordinator/internal/storage/postgres/migrations/0014_workload_settings.down.sql create mode 100644 coordinator/internal/storage/postgres/migrations/0014_workload_settings.up.sql create mode 100644 coordinator/internal/storage/postgres/workload_settings_repo.go create mode 100644 coordinator/internal/storage/sqlite/admin_m2_test.go create mode 100644 coordinator/internal/storage/sqlite/migrations/0002_workload_settings.sql create mode 100644 coordinator/internal/storage/sqlite/workload_settings_repo.go create mode 100644 coordinator/internal/userservice/usecase/users.go diff --git a/coordinator/cmd/coordinator/main.go b/coordinator/cmd/coordinator/main.go index 965fece..90d352f 100644 --- a/coordinator/cmd/coordinator/main.go +++ b/coordinator/cmd/coordinator/main.go @@ -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) }, diff --git a/coordinator/internal/domain/errors.go b/coordinator/internal/domain/errors.go index fca2979..7e8a825 100644 --- a/coordinator/internal/domain/errors.go +++ b/coordinator/internal/domain/errors.go @@ -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") ) diff --git a/coordinator/internal/memstore/memstore.go b/coordinator/internal/memstore/memstore.go index 9c911e4..ace397f 100644 --- a/coordinator/internal/memstore/memstore.go +++ b/coordinator/internal/memstore/memstore.go @@ -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 { diff --git a/coordinator/internal/storage/postgres/migrate_test.go b/coordinator/internal/storage/postgres/migrate_test.go index 7b241f2..f7f3789 100644 --- a/coordinator/internal/storage/postgres/migrate_test.go +++ b/coordinator/internal/storage/postgres/migrate_test.go @@ -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 "" } diff --git a/coordinator/internal/storage/postgres/migrations/0014_workload_settings.down.sql b/coordinator/internal/storage/postgres/migrations/0014_workload_settings.down.sql new file mode 100644 index 0000000..b79ba91 --- /dev/null +++ b/coordinator/internal/storage/postgres/migrations/0014_workload_settings.down.sql @@ -0,0 +1,5 @@ +BEGIN; + +DROP TABLE IF EXISTS workload_settings; + +COMMIT; diff --git a/coordinator/internal/storage/postgres/migrations/0014_workload_settings.up.sql b/coordinator/internal/storage/postgres/migrations/0014_workload_settings.up.sql new file mode 100644 index 0000000..2975b96 --- /dev/null +++ b/coordinator/internal/storage/postgres/migrations/0014_workload_settings.up.sql @@ -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; diff --git a/coordinator/internal/storage/postgres/worker_repo.go b/coordinator/internal/storage/postgres/worker_repo.go index 5f75183..e3d2bc5 100644 --- a/coordinator/internal/storage/postgres/worker_repo.go +++ b/coordinator/internal/storage/postgres/worker_repo.go @@ -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 diff --git a/coordinator/internal/storage/postgres/workload_settings_repo.go b/coordinator/internal/storage/postgres/workload_settings_repo.go new file mode 100644 index 0000000..f668f56 --- /dev/null +++ b/coordinator/internal/storage/postgres/workload_settings_repo.go @@ -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 +} diff --git a/coordinator/internal/storage/sqlite/admin_m2_test.go b/coordinator/internal/storage/sqlite/admin_m2_test.go new file mode 100644 index 0000000..ef3a060 --- /dev/null +++ b/coordinator/internal/storage/sqlite/admin_m2_test.go @@ -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) + } +} diff --git a/coordinator/internal/storage/sqlite/migrations/0002_workload_settings.sql b/coordinator/internal/storage/sqlite/migrations/0002_workload_settings.sql new file mode 100644 index 0000000..7c3b7a8 --- /dev/null +++ b/coordinator/internal/storage/sqlite/migrations/0002_workload_settings.sql @@ -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 +); diff --git a/coordinator/internal/storage/sqlite/sqlite_test.go b/coordinator/internal/storage/sqlite/sqlite_test.go index 78eb679..55792b5 100644 --- a/coordinator/internal/storage/sqlite/sqlite_test.go +++ b/coordinator/internal/storage/sqlite/sqlite_test.go @@ -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) } } diff --git a/coordinator/internal/storage/sqlite/worker_repo.go b/coordinator/internal/storage/sqlite/worker_repo.go index 4d0bf44..3dbfc17 100644 --- a/coordinator/internal/storage/sqlite/worker_repo.go +++ b/coordinator/internal/storage/sqlite/worker_repo.go @@ -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 +} diff --git a/coordinator/internal/storage/sqlite/workload_settings_repo.go b/coordinator/internal/storage/sqlite/workload_settings_repo.go new file mode 100644 index 0000000..c41c53a --- /dev/null +++ b/coordinator/internal/storage/sqlite/workload_settings_repo.go @@ -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 +} diff --git a/coordinator/internal/transport/http/errors.go b/coordinator/internal/transport/http/errors.go index ae14782..ef8147b 100644 --- a/coordinator/internal/transport/http/errors.go +++ b/coordinator/internal/transport/http/errors.go @@ -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): diff --git a/coordinator/internal/transport/http/server.go b/coordinator/internal/transport/http/server.go index d01edd0..b919846 100644 --- a/coordinator/internal/transport/http/server.go +++ b/coordinator/internal/transport/http/server.go @@ -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) diff --git a/coordinator/internal/transport/http/server_test.go b/coordinator/internal/transport/http/server_test.go index fd6b2f8..3c401b6 100644 --- a/coordinator/internal/transport/http/server_test.go +++ b/coordinator/internal/transport/http/server_test.go @@ -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 +} diff --git a/coordinator/internal/transport/http/templates/admin.html b/coordinator/internal/transport/http/templates/admin.html index ba0f603..e920f57 100644 --- a/coordinator/internal/transport/http/templates/admin.html +++ b/coordinator/internal/transport/http/templates/admin.html @@ -205,13 +205,35 @@ tbody tr:hover{background:var(--panel-2)} - +
-

Workers

milestone M2
Trust management lands with milestone M2Worker list, trust dropdown and heartbeat overview are wired next. The dashboard already shows the live fleet.
+
Workers register themselves. Trust decides whether a machine's results are accepted directly or need quorum.
+
+ + + +
WorkerStatusCapabilitiesTrustOwnerLast signal
+
- +
+
Users
+
+ + + +
EmailRoleVerifiedCreated
+
—
+
+
Worker keys
+
Keys let lab machines register as workers under a user account. Served instances can also use the cluster token.
+
+ + + +
LabelPrefixOwnerCreatedLast used
+
Quick user action
@@ -226,13 +248,17 @@ tbody tr:hover{background:var(--panel-2)} {{if .Error}}
✗ {{.Error}}
{{end}}
-
Accounts and worker keys
-
User list and key management land with milestone M2The userservice gains admin list endpoints; the console gets tables, role selects and key revoke.
- +
-
Workload enable/disable lands with milestone M2The catalog is already served to the job form; persisted on/off switches arrive with the settings migration.
+
Disabled workloads are rejected at submit time and hidden from the job form. Settings persist in the database.
+
+ + + +
WorkloadReductionParametersDataset uploadEnabled
+
@@ -255,9 +281,20 @@ tbody tr:hover{background:var(--panel-2)} - +
-
Cluster settings land with milestone M2Worker token reveal, public URL and storage settings follow in the next milestone.
+
The cluster token below authenticates any worker. Reveal it only on a trusted machine.
+
Cluster
+
+
+
Worker token
••••••••••••••••••••••••
+
Public URL
—
+
Listen address
—
+
Data directory
—
+
Database engine
—
+
Binary
—
+
+
@@ -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=>({'&':'&','<':'<','>':'>','"':'"'}[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='
No worker is registered yet.
';rows.append(tr);return} + for(const w of v.workers){ + const tr=document.createElement('tr'); + const [label,cls]=workerStatusPill(w.status); + const trustSel=''; + tr.innerHTML='
'+esc(w.name)+'
'+esc(w.id.slice(0,8))+'…
'+ + ''+pill(label,cls,null)+''+ + ''+(w.capabilities||[]).map(c=>''+esc(c)+'').join('')+''+ + ''+trustSel+''+ + ''+esc(w.owner)+''+ + ''+fmtTime(w.last_heartbeat_at)+''; + 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=''; + tr.innerHTML='
'+esc(u.email)+'
'+ + ''+roleSel+''+ + ''+pill(u.verified?'Verified':'—',u.verified?'pill-success':'pill-waiting',null)+''+ + ''+fmtTime(u.created_at)+''; + 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=''+esc(k.name)+''+ + ''+esc(k.prefix)+'…'+ + ''+esc(emailOf[k.user_id]||k.user_id.slice(0,8)+'…')+''+ + ''+fmtTime(k.created_at)+''+ + ''+(k.last_used_at?fmtTime(k.last_used_at):'never')+''+ + ''+((k.revoked_at)?'Revoked':'')+''; + 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='
'+esc(w.name)+'
'+esc(w.description||'')+'
'+ + ''+esc(w.reduction)+''+ + ''+w.parameters+' declared'+ + ''+pill(w.upload_ready?'ready':'—',w.upload_ready?'pill-success':'pill-waiting',null)+''+ + ''; + 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'))); diff --git a/coordinator/internal/transport/http/ui_admin.go b/coordinator/internal/transport/http/ui_admin.go index c20f9ba..29d2f14 100644 --- a/coordinator/internal/transport/http/ui_admin.go +++ b/coordinator/internal/transport/http/ui_admin.go @@ -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. diff --git a/coordinator/internal/usecase/admin.go b/coordinator/internal/usecase/admin.go index 9613a0e..0e2a233 100644 --- a/coordinator/internal/usecase/admin.go +++ b/coordinator/internal/usecase/admin.go @@ -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 +} diff --git a/coordinator/internal/usecase/admin_test.go b/coordinator/internal/usecase/admin_test.go index 9b85cf5..19ff436 100644 --- a/coordinator/internal/usecase/admin_test.go +++ b/coordinator/internal/usecase/admin_test.go @@ -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) + } +} diff --git a/coordinator/internal/usecase/ports.go b/coordinator/internal/usecase/ports.go index d903816..1798529 100644 --- a/coordinator/internal/usecase/ports.go +++ b/coordinator/internal/usecase/ports.go @@ -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") diff --git a/coordinator/internal/usecase/upload.go b/coordinator/internal/usecase/upload.go index 8fa4798..b273933 100644 --- a/coordinator/internal/usecase/upload.go +++ b/coordinator/internal/usecase/upload.go @@ -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 } diff --git a/coordinator/internal/usecase/usecase_test.go b/coordinator/internal/usecase/usecase_test.go index 842eca9..b1e29c3 100644 --- a/coordinator/internal/usecase/usecase_test.go +++ b/coordinator/internal/usecase/usecase_test.go @@ -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) + } +} diff --git a/coordinator/internal/userservice/embedded.go b/coordinator/internal/userservice/embedded.go index 6c7c983..1021d47 100644 --- a/coordinator/internal/userservice/embedded.go +++ b/coordinator/internal/userservice/embedded.go @@ -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 != "" { diff --git a/coordinator/internal/userservice/memstore/memstore.go b/coordinator/internal/userservice/memstore/memstore.go index ed7c80e..dbf8594 100644 --- a/coordinator/internal/userservice/memstore/memstore.go +++ b/coordinator/internal/userservice/memstore/memstore.go @@ -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, ©) + } + 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 +} diff --git a/coordinator/internal/userservice/storage/sqlite/sqlite.go b/coordinator/internal/userservice/storage/sqlite/sqlite.go index 65c2815..4eecc6a 100644 --- a/coordinator/internal/userservice/storage/sqlite/sqlite.go +++ b/coordinator/internal/userservice/storage/sqlite/sqlite.go @@ -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 = ?", diff --git a/coordinator/internal/userservice/transport/http/dto.go b/coordinator/internal/userservice/transport/http/dto.go index 43cc750..b6c2e36 100644 --- a/coordinator/internal/userservice/transport/http/dto.go +++ b/coordinator/internal/userservice/transport/http/dto.go @@ -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(), diff --git a/coordinator/internal/userservice/transport/http/handlers.go b/coordinator/internal/userservice/transport/http/handlers.go index 5b70266..c2a421c 100644 --- a/coordinator/internal/userservice/transport/http/handlers.go +++ b/coordinator/internal/userservice/transport/http/handlers.go @@ -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 diff --git a/coordinator/internal/userservice/transport/http/server.go b/coordinator/internal/userservice/transport/http/server.go index 6f5df0b..793cc36 100644 --- a/coordinator/internal/userservice/transport/http/server.go +++ b/coordinator/internal/userservice/transport/http/server.go @@ -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)) } diff --git a/coordinator/internal/userservice/transport/http/server_test.go b/coordinator/internal/userservice/transport/http/server_test.go index 19b6586..a1c7699 100644 --- a/coordinator/internal/userservice/transport/http/server_test.go +++ b/coordinator/internal/userservice/transport/http/server_test.go @@ -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) + } +} diff --git a/coordinator/internal/userservice/usecase/errorpaths_test.go b/coordinator/internal/userservice/usecase/errorpaths_test.go index 573d3ac..208af9f 100644 --- a/coordinator/internal/userservice/usecase/errorpaths_test.go +++ b/coordinator/internal/userservice/usecase/errorpaths_test.go @@ -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 diff --git a/coordinator/internal/userservice/usecase/ports.go b/coordinator/internal/userservice/usecase/ports.go index 047e9cd..e196a1e 100644 --- a/coordinator/internal/userservice/usecase/ports.go +++ b/coordinator/internal/userservice/usecase/ports.go @@ -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 diff --git a/coordinator/internal/userservice/usecase/users.go b/coordinator/internal/userservice/usecase/users.go new file mode 100644 index 0000000..f4e23ce --- /dev/null +++ b/coordinator/internal/userservice/usecase/users.go @@ -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) +} diff --git a/coordinator/internal/userservice/usecase/workerkey.go b/coordinator/internal/userservice/usecase/workerkey.go index 1551203..f189dd1 100644 --- a/coordinator/internal/userservice/usecase/workerkey.go +++ b/coordinator/internal/userservice/usecase/workerkey.go @@ -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 diff --git a/coordinator/internal/userservice/usecase/workerkey_test.go b/coordinator/internal/userservice/usecase/workerkey_test.go index 75ecfc2..8b3684e 100644 --- a/coordinator/internal/userservice/usecase/workerkey_test.go +++ b/coordinator/internal/userservice/usecase/workerkey_test.go @@ -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() diff --git a/coordinator/internal/workloads/catalog.go b/coordinator/internal/workloads/catalog.go index 2dbcb86..77c1c6a 100644 --- a/coordinator/internal/workloads/catalog.go +++ b/coordinator/internal/workloads/catalog.go @@ -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 {