Compare commits

...
3 Commits
Author SHA1 Message Date
tao.chenandClaude df9b01272f admin: uploads list view with search, sort, and delete
Add /admin/uploads HTML page and JSON API backed by the upload_files

DB index. The page lists file_id, name, size (KB), uploaded_at, and

sha256 with client-side search by name, column sort, and per-row delete.

DELETE /admin/uploads/:id removes the DB index row and the on-disk

data file plus .meta.json sidecar. Admin deletes are written to the

audit log.

Wire the new dependencies through admin.Mount and main.go, and run

the backfill step on startup so pre-existing sidecars are indexed.

Tests cover auth, HTML rendering, JSON search, delete removing all

artifacts, and audit entry creation.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-14 11:39:56 +08:00
tao.chenandClaude 876f1c9edb uploads: dual-write uploads to DB, startup backfill, and audit log
Store now indexes every Save into the upload_files table via SetRepo.

DB failures are logged as warnings and do not fail the upload because

.meta.json remains the source of truth and startup backfill recovers.

Add Store.Backfill to walk Root at startup and insert index rows for

any pre-existing .meta.json sidecars, swallowing duplicate-key races.

The upload_file MCP Tool now writes an audit_log entry on success.

Tests cover dual-write args, repo-error non-failure, and backfill

skipping existing rows.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-14 11:30:39 +08:00
tao.chenandClaude 1796b5b872 storage: add upload_files table and UploadRepo
Add the upload_files SQLite table and UploadRepo CRUD methods

(Create/Get/List/Delete). The table stores file_id, name, size,

sha256, and uploaded_at as a queryable index.

The .meta.json sidecar remains the source of truth for uploads;

the DB is only an index for admin visibility. On startup the next

commit will backfill any pre-existing sidecars into this table.

Tests cover Create/Get/List/Delete round-trip, name search

filtering, descending time ordering, and limit defaults/cap.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-14 11:21:38 +08:00
17 changed files with 1275 additions and 22 deletions
+67 -1
View File
@@ -7,6 +7,8 @@ import (
"errors" "errors"
"log/slog" "log/slog"
"net/http" "net/http"
"os"
"path/filepath"
"strconv" "strconv"
"github.com/gin-gonic/gin" "github.com/gin-gonic/gin"
@@ -15,11 +17,12 @@ import (
"spark-mcp-go/internal/cluster" "spark-mcp-go/internal/cluster"
"spark-mcp-go/internal/middleware" "spark-mcp-go/internal/middleware"
"spark-mcp-go/internal/storage" "spark-mcp-go/internal/storage"
"spark-mcp-go/internal/uploads"
) )
// Mount attaches the /admin sub-router to r, protecting every route with // Mount attaches the /admin sub-router to r, protecting every route with
// bearer-token admin authentication. // bearer-token admin authentication.
func Mount(r *gin.Engine, repo *storage.ClusterRepo, auditRepo *audit.Repo, adminTokens []string) { func Mount(r *gin.Engine, repo *storage.ClusterRepo, uploadRepo *storage.UploadRepo, auditRepo *audit.Repo, uploadStore *uploads.Store, adminTokens []string) {
// HTML page is public — its own modal prompts for the token. // HTML page is public — its own modal prompts for the token.
// All other /admin/* endpoints (API + OpenAPI docs) still require auth. // All other /admin/* endpoints (API + OpenAPI docs) still require auth.
gPublic := r.Group("/admin") gPublic := r.Group("/admin")
@@ -33,6 +36,9 @@ func Mount(r *gin.Engine, repo *storage.ClusterRepo, auditRepo *audit.Repo, admi
g.PUT("/clusters/:id", updateCluster(repo, auditRepo)) g.PUT("/clusters/:id", updateCluster(repo, auditRepo))
g.DELETE("/clusters/:id", deleteCluster(repo, auditRepo)) g.DELETE("/clusters/:id", deleteCluster(repo, auditRepo))
g.GET("/audit", listAudit(auditRepo)) g.GET("/audit", listAudit(auditRepo))
g.GET("/uploads", uploadsWebHandler)
g.GET("/uploads/api", listUploads(uploadRepo))
g.DELETE("/uploads/:id", deleteUpload(uploadRepo, auditRepo, uploadStore))
g.GET("/docs", DocsHandler) g.GET("/docs", DocsHandler)
g.GET("/docs/spec", OpenAPISpecHandler) g.GET("/docs/spec", OpenAPISpecHandler)
} }
@@ -275,6 +281,66 @@ func deleteCluster(repo *storage.ClusterRepo, auditRepo *audit.Repo) gin.Handler
} }
} }
func listUploads(repo *storage.UploadRepo) gin.HandlerFunc {
return func(c *gin.Context) {
search := c.Query("search")
limitStr := c.DefaultQuery("limit", "100")
limit, err := strconv.Atoi(limitStr)
if err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid limit"})
return
}
uploads, err := repo.List(c.Request.Context(), search, limit)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, uploads)
}
}
func deleteUpload(repo *storage.UploadRepo, auditRepo *audit.Repo, store *uploads.Store) gin.HandlerFunc {
return func(c *gin.Context) {
id := c.Param("id")
ctx := c.Request.Context()
meta, err := repo.Get(ctx, id)
if err != nil && !errors.Is(err, storage.ErrNotFound) {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
if err := repo.Delete(ctx, id); err != nil && !errors.Is(err, storage.ErrNotFound) {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
// Always attempt to remove the on-disk files; the DB is only an index.
if store != nil {
dataPath := filepath.Join(store.Root, id)
_ = os.Remove(dataPath)
_ = os.Remove(dataPath + ".meta.json")
}
if meta.FileID != "" {
details, _ := audit.MarshalDetails(map[string]any{
"file_id": id,
"name": meta.Name,
"size": meta.Size,
"sha256": meta.SHA256,
})
_ = auditRepo.Insert(ctx, &audit.Entry{
Actor: actor(c),
Action: audit.ActionUploadDelete,
ClusterID: id,
Details: details,
})
}
c.Status(http.StatusNoContent)
}
}
func listAudit(auditRepo *audit.Repo) gin.HandlerFunc { func listAudit(auditRepo *audit.Repo) gin.HandlerFunc {
return func(c *gin.Context) { return func(c *gin.Context) {
limitStr := c.DefaultQuery("limit", "100") limitStr := c.DefaultQuery("limit", "100")
+7 -1
View File
@@ -7,6 +7,7 @@ import (
"io" "io"
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"path/filepath"
"strings" "strings"
"testing" "testing"
@@ -15,6 +16,7 @@ import (
"spark-mcp-go/internal/audit" "spark-mcp-go/internal/audit"
"spark-mcp-go/internal/cluster" "spark-mcp-go/internal/cluster"
"spark-mcp-go/internal/storage" "spark-mcp-go/internal/storage"
"spark-mcp-go/internal/uploads"
) )
func init() { func init() {
@@ -29,7 +31,11 @@ func newTestAdmin(t *testing.T) (*gin.Engine, *storage.ClusterRepo, *audit.Repo)
} }
t.Cleanup(func() { _ = db.Close() }) t.Cleanup(func() { _ = db.Close() })
r := gin.New() r := gin.New()
Mount(r, db.Clusters(), audit.NewRepo(db), []string{"good-token"}) uploadStore, err := uploads.New(filepath.Join(t.TempDir(), "uploads"))
if err != nil {
t.Fatalf("new upload store: %v", err)
}
Mount(r, db.Clusters(), db.Uploads(), audit.NewRepo(db), &uploadStore, []string{"good-token"})
return r, db.Clusters(), audit.NewRepo(db) return r, db.Clusters(), audit.NewRepo(db)
} }
+154
View File
@@ -0,0 +1,154 @@
package admin
import (
"context"
"encoding/json"
"errors"
"net/http"
"os"
"path/filepath"
"strings"
"testing"
"time"
"github.com/gin-gonic/gin"
"spark-mcp-go/internal/audit"
"spark-mcp-go/internal/storage"
"spark-mcp-go/internal/uploads"
)
func newTestAdminUploads(t *testing.T) (*gin.Engine, *storage.UploadRepo, *audit.Repo, *uploads.Store) {
t.Helper()
db, err := storage.Open(":memory:")
if err != nil {
t.Fatalf("open: %v", err)
}
t.Cleanup(func() { _ = db.Close() })
uploadStore, err := uploads.New(filepath.Join(t.TempDir(), "uploads"))
if err != nil {
t.Fatalf("new upload store: %v", err)
}
r := gin.New()
Mount(r, db.Clusters(), db.Uploads(), audit.NewRepo(db), &uploadStore, []string{"good-token"})
return r, db.Uploads(), audit.NewRepo(db), &uploadStore
}
func TestUploadsWeb_RequiresAuth(t *testing.T) {
r, _, _, _ := newTestAdminUploads(t)
w := doReq(t, r, "GET", "/admin/uploads", "", nil)
if w.Code != http.StatusUnauthorized {
t.Errorf("GET /admin/uploads without auth: got %d, want %d", w.Code, http.StatusUnauthorized)
}
w = doReq(t, r, "GET", "/admin/uploads", "good-token", nil)
if w.Code != http.StatusOK {
t.Errorf("GET /admin/uploads with auth: got %d, want %d", w.Code, http.StatusOK)
}
ct := w.Header().Get("Content-Type")
if !strings.Contains(ct, "text/html") {
t.Errorf("Content-Type = %q, want text/html", ct)
}
body := w.Body.String()
if !strings.Contains(body, "href='/admin/uploads'") && !strings.Contains(body, "href=\"/admin/uploads\"") {
t.Errorf("body missing uploads nav link")
}
if !strings.Contains(body, "uploads-body") {
t.Errorf("body missing uploads table")
}
}
func TestListUploads(t *testing.T) {
r, repo, _, _ := newTestAdminUploads(t)
ctx := context.Background()
uploadedAt := time.Unix(0, time.Now().UnixNano())
if err := repo.Create(ctx, "00000000000000000000000000000001", "alpha.txt", 10, "deadbeef", uploadedAt); err != nil {
t.Fatalf("create upload: %v", err)
}
if err := repo.Create(ctx, "00000000000000000000000000000002", "beta.txt", 20, "cafebabe", uploadedAt.Add(time.Second)); err != nil {
t.Fatalf("create upload: %v", err)
}
w := doReq(t, r, "GET", "/admin/uploads/api", "good-token", nil)
if w.Code != http.StatusOK {
t.Fatalf("got status %d, want %d", w.Code, http.StatusOK)
}
var list []map[string]any
if err := json.Unmarshal(w.Body.Bytes(), &list); err != nil {
t.Fatalf("unmarshal list: %v", err)
}
if len(list) != 2 {
t.Errorf("got %d uploads, want 2", len(list))
}
w = doReq(t, r, "GET", "/admin/uploads/api?search=alpha", "good-token", nil)
if w.Code != http.StatusOK {
t.Fatalf("search: got status %d", w.Code)
}
if err := json.Unmarshal(w.Body.Bytes(), &list); err != nil {
t.Fatalf("unmarshal search: %v", err)
}
if len(list) != 1 {
t.Errorf("search result: got %d, want 1", len(list))
}
}
func TestDeleteUpload(t *testing.T) {
r, repo, auditRepo, store := newTestAdminUploads(t)
ctx := context.Background()
id := "00000000000000000000000000000003"
dataPath := filepath.Join(store.Root, id)
metaPath := dataPath + ".meta.json"
if err := os.WriteFile(dataPath, []byte("payload"), 0o640); err != nil {
t.Fatalf("create data file: %v", err)
}
sc := map[string]any{
"name": "delete-me.txt",
"size": 7,
"sha256": "feedface",
"uploaded_at": time.Now().Format(time.RFC3339Nano),
}
b, _ := json.Marshal(sc)
if err := os.WriteFile(metaPath, b, 0o600); err != nil {
t.Fatalf("create sidecar: %v", err)
}
if err := repo.Create(ctx, id, "delete-me.txt", 7, "feedface", time.Now()); err != nil {
t.Fatalf("create db row: %v", err)
}
w := doReq(t, r, "DELETE", "/admin/uploads/"+id, "good-token", nil)
if w.Code != http.StatusNoContent {
t.Fatalf("delete: got status %d, want %d: %s", w.Code, http.StatusNoContent, w.Body.String())
}
if _, err := repo.Get(ctx, id); !errors.Is(err, storage.ErrNotFound) {
t.Errorf("db row still exists: %v", err)
}
if _, err := os.Stat(dataPath); !os.IsNotExist(err) {
t.Errorf("data file still exists")
}
if _, err := os.Stat(metaPath); !os.IsNotExist(err) {
t.Errorf("sidecar still exists")
}
entries, err := auditRepo.List(ctx, 100)
if err != nil {
t.Fatalf("list audit: %v", err)
}
found := false
for _, e := range entries {
if e.Action == audit.ActionUploadDelete && e.ClusterID == id {
found = true
break
}
}
if !found {
t.Errorf("missing upload.delete audit entry")
}
}
+10 -1
View File
@@ -7,7 +7,7 @@ import (
"github.com/gin-gonic/gin" "github.com/gin-gonic/gin"
) )
//go:embed web/cluster.html //go:embed web
var webFS embed.FS var webFS embed.FS
func webHandler(c *gin.Context) { func webHandler(c *gin.Context) {
@@ -18,3 +18,12 @@ func webHandler(c *gin.Context) {
} }
c.Data(http.StatusOK, "text/html; charset=utf-8", data) c.Data(http.StatusOK, "text/html; charset=utf-8", data)
} }
func uploadsWebHandler(c *gin.Context) {
data, err := webFS.ReadFile("web/uploads.html")
if err != nil {
c.String(http.StatusInternalServerError, "uploads.html: %v", err)
return
}
c.Data(http.StatusOK, "text/html; charset=utf-8", data)
}
+10
View File
@@ -34,6 +34,10 @@
flex-wrap: wrap; flex-wrap: wrap;
} }
header h1 { margin: 0; font-size: 1.1rem; color: var(--accent); } header h1 { margin: 0; font-size: 1.1rem; color: var(--accent); }
nav { display: flex; gap: .75rem; }
nav a { color: var(--muted); text-decoration: none; }
nav a:hover { color: var(--text); }
nav a.active { color: var(--accent); }
.auth { display: flex; align-items: center; gap: .5rem; flex-wrap: wrap; } .auth { display: flex; align-items: center; gap: .5rem; flex-wrap: wrap; }
input[type='text'], input[type='password'], input[type='number'], select, textarea { input[type='text'], input[type='password'], input[type='number'], select, textarea {
background: var(--bg); background: var(--bg);
@@ -111,7 +115,13 @@
</head> </head>
<body> <body>
<header> <header>
<div style='display:flex;align-items:center;gap:1rem;'>
<h1>spark-mcp-go Admin</h1> <h1>spark-mcp-go Admin</h1>
<nav>
<a href='/admin' class='active'>Clusters</a>
<a href='/admin/uploads'>Uploads</a>
</nav>
</div>
<div class='auth'> <div class='auth'>
<button id='set-token' class='secondary'>Set Token</button> <button id='set-token' class='secondary'>Set Token</button>
<button id='logout' class='danger'>Logout</button> <button id='logout' class='danger'>Logout</button>
+377
View File
@@ -0,0 +1,377 @@
<!DOCTYPE html>
<html lang='en'>
<head>
<meta charset='utf-8'>
<meta name='viewport' content='width=device-width, initial-scale=1'>
<title>Uploads - spark-mcp-go Admin</title>
<style>
:root {
--bg: #0d1117;
--panel: #161b22;
--border: #30363d;
--text: #c9d1d9;
--muted: #8b949e;
--accent: #58a6ff;
--danger: #f85149;
--ok: #3fb950;
}
* { box-sizing: border-box; }
body {
margin: 0;
font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, 'Liberation Mono', monospace;
background: var(--bg);
color: var(--text);
line-height: 1.5;
}
header {
display: flex;
align-items: center;
justify-content: space-between;
gap: 1rem;
padding: 1rem;
border-bottom: 1px solid var(--border);
background: var(--panel);
flex-wrap: wrap;
}
header h1 { margin: 0; font-size: 1.1rem; color: var(--accent); }
nav { display: flex; gap: .75rem; }
nav a { color: var(--muted); text-decoration: none; }
nav a:hover { color: var(--text); }
nav a.active { color: var(--accent); }
.auth { display: flex; align-items: center; gap: .5rem; flex-wrap: wrap; }
input[type='text'], input[type='password'] {
background: var(--bg);
color: var(--text);
border: 1px solid var(--border);
border-radius: 4px;
padding: .4rem .5rem;
font: inherit;
}
input[type='text'] { min-width: 220px; }
button {
background: var(--accent);
color: #fff;
border: none;
border-radius: 4px;
padding: .4rem .8rem;
font: inherit;
cursor: pointer;
}
button:hover { opacity: .9; }
button.danger { background: var(--danger); }
button.secondary { background: var(--border); color: var(--text); }
main { padding: 1rem; max-width: 1200px; margin: 0 auto; }
.toolbar { display: flex; gap: .5rem; margin-bottom: 1rem; align-items: center; flex-wrap: wrap; }
.toolbar span { color: var(--muted); }
.error {
background: rgba(248, 81, 73, .15);
border: 1px solid var(--danger);
color: var(--danger);
padding: .75rem;
border-radius: 4px;
margin-bottom: 1rem;
white-space: pre-wrap;
}
.hidden { display: none; }
table { width: 100%; border-collapse: collapse; }
th, td { text-align: left; padding: .5rem; border-bottom: 1px solid var(--border); }
th { color: var(--muted); font-weight: 600; user-select: none; }
th.sortable { cursor: pointer; }
th.sortable:hover { color: var(--text); }
td { vertical-align: middle; font-size: .9rem; }
td.mono { font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, monospace; }
.actions { display: flex; gap: .4rem; }
dialog.modal {
border: 1px solid var(--border);
background: var(--panel);
color: var(--text);
border-radius: 6px;
padding: 1.5rem;
min-width: 360px;
max-width: 90vw;
}
dialog.modal::backdrop { background: rgba(0, 0, 0, .65); }
dialog.modal h2 { margin: 0 0 .5rem; font-size: 1.1rem; color: var(--accent); }
dialog.modal form { display: flex; flex-direction: column; gap: .75rem; }
dialog.modal label { color: var(--muted); font-size: .9rem; }
dialog.modal input { width: 100%; }
.modal-error { color: var(--danger); font-size: .9rem; margin: 0; }
.modal-actions { display: flex; gap: .5rem; justify-content: flex-end; }
</style>
</head>
<body>
<header>
<div style='display:flex;align-items:center;gap:1rem;'>
<h1>spark-mcp-go Admin</h1>
<nav>
<a href='/admin'>Clusters</a>
<a href='/admin/uploads' class='active'>Uploads</a>
</nav>
</div>
<div class='auth'>
<button id='set-token' class='secondary'>Set Token</button>
<button id='logout' class='danger'>Logout</button>
</div>
</header>
<main>
<div id='error' class='error hidden'></div>
<section class='toolbar'>
<input id='search' type='text' placeholder='Search by name...'>
<button id='refresh' class='secondary'>Refresh</button>
<span id='count'></span>
</section>
<section>
<table>
<thead>
<tr>
<th class='sortable' data-key='file_id'>File ID</th>
<th class='sortable' data-key='name'>Name</th>
<th class='sortable' data-key='size'>Size (KB)</th>
<th class='sortable' data-key='uploaded_at'>Uploaded At</th>
<th>SHA256</th>
<th>Actions</th>
</tr>
</thead>
<tbody id='uploads-body'>
<tr><td colspan='6'>Loading...</td></tr>
</tbody>
</table>
</section>
</main>
<dialog id='token-modal' class='modal'>
<h2>Admin Token Required</h2>
<p id='modal-error' class='modal-error hidden'></p>
<form id='token-form'>
<label for='modal-token'>Bearer token</label>
<input id='modal-token' type='password' autocomplete='off' required>
<div class='modal-actions'>
<button type='submit' id='modal-save'>Save</button>
<button type='button' id='modal-cancel' class='secondary hidden'>Cancel</button>
</div>
</form>
</dialog>
<script>
const $ = (sel) => document.querySelector(sel);
const API = '/admin/uploads';
let uploads = [];
let sortKey = 'uploaded_at';
let sortDir = -1;
let pendingRetry = null;
let cancelBlocker = null;
function headers() {
return {
'Authorization': 'Bearer ' + (localStorage.getItem('adminToken') || ''),
'Content-Type': 'application/json'
};
}
function showError(msg) {
const el = $('#error');
el.textContent = msg || '';
el.classList.toggle('hidden', !msg);
}
function showModalError(msg) {
const el = $('#modal-error');
el.textContent = msg || '';
el.classList.toggle('hidden', !msg);
}
function openTokenModal(opts = {}) {
const dialog = $('#token-modal');
const cancelBtn = $('#modal-cancel');
showModalError(opts.error || '');
cancelBtn.classList.toggle('hidden', !opts.allowCancel);
if (cancelBlocker) {
dialog.removeEventListener('cancel', cancelBlocker);
cancelBlocker = null;
}
if (!opts.allowCancel) {
cancelBlocker = (e) => e.preventDefault();
dialog.addEventListener('cancel', cancelBlocker);
}
dialog.showModal();
$('#modal-token').focus();
}
function closeTokenModal() {
const dialog = $('#token-modal');
showModalError('');
if (cancelBlocker) {
dialog.removeEventListener('cancel', cancelBlocker);
cancelBlocker = null;
}
dialog.close();
$('#modal-token').value = '';
}
function saveTokenFromModal(e) {
e.preventDefault();
localStorage.setItem('adminToken', $('#modal-token').value.trim());
closeTokenModal();
if (pendingRetry) {
const fn = pendingRetry;
pendingRetry = null;
fn();
} else {
loadUploads();
}
}
async function api(method, path, body) {
const opts = { method, headers: headers() };
if (body !== undefined) opts.body = JSON.stringify(body);
const res = await fetch(API + path, opts);
if (!res.ok) {
if (res.status === 401) {
localStorage.removeItem('adminToken');
pendingRetry = () => api(method, path, body);
openTokenModal({
error: 'Token rejected by server (401). Enter a valid token.',
allowCancel: false
});
throw new Error('401 Unauthorized');
}
let detail = res.statusText;
try {
const j = await res.json();
if (j.error) detail = j.error;
} catch (_) {}
throw new Error(`${res.status} ${detail}`);
}
if (res.status === 204) return null;
return res.json();
}
function formatKB(bytes) {
return (bytes / 1024).toFixed(2);
}
function formatTime(ts) {
if (!ts) return '';
const d = new Date(ts);
return isNaN(d) ? ts : d.toISOString();
}
function truncate(s, n) {
if (!s || s.length <= n) return s || '';
return s.slice(0, n) + '…';
}
function escapeHtml(s) {
return String(s)
.replace(/&/g, '&amp;')
.replace(/</g, '&lt;')
.replace(/>/g, '&gt;')
.replace(/"/g, '&quot;')
.replace(/'/g, '&#39;');
}
function filteredUploads() {
const q = $('#search').value.trim().toLowerCase();
if (!q) return uploads.slice();
return uploads.filter(u => (u.name || '').toLowerCase().includes(q));
}
function compare(a, b) {
let av = a[sortKey];
let bv = b[sortKey];
if (sortKey === 'size') { av = Number(av); bv = Number(bv); }
else if (sortKey === 'uploaded_at') { av = new Date(av).getTime(); bv = new Date(bv).getTime(); }
else { av = String(av).toLowerCase(); bv = String(bv).toLowerCase(); }
if (av < bv) return -sortDir;
if (av > bv) return sortDir;
return 0;
}
function renderList() {
const tbody = $('#uploads-body');
const list = filteredUploads().sort(compare);
$('#count').textContent = list.length + ' file' + (list.length === 1 ? '' : 's');
tbody.innerHTML = '';
if (!list.length) {
tbody.innerHTML = '<tr><td colspan="6">No uploads.</td></tr>';
return;
}
for (const u of list) {
const tr = document.createElement('tr');
tr.innerHTML = `
<td class='mono' title='${escapeHtml(u.file_id)}'>${escapeHtml(truncate(u.file_id, 8))}</td>
<td>${escapeHtml(u.name)}</td>
<td>${escapeHtml(formatKB(u.size))}</td>
<td>${escapeHtml(formatTime(u.uploaded_at))}</td>
<td class='mono' title='${escapeHtml(u.sha256)}'>${escapeHtml(truncate(u.sha256, 8))}</td>
<td class='actions'>
<button data-id='${escapeHtml(u.file_id)}' class='danger delete'>Delete</button>
</td>
`;
tbody.appendChild(tr);
}
tbody.querySelectorAll('.delete').forEach(b => b.addEventListener('click', handleDelete));
}
async function loadUploads() {
showError('');
try {
uploads = await api('GET', '/api');
renderList();
} catch (e) {
showError('Failed to load uploads: ' + e.message);
}
}
async function handleDelete(e) {
const id = e.target.dataset.id;
if (!confirm(`Delete upload ${id}?`)) return;
showError('');
try {
await api('DELETE', `/${encodeURIComponent(id)}`);
await loadUploads();
} catch (err) {
showError('Failed to delete upload: ' + err.message);
}
}
function handleSort(e) {
const key = e.target.dataset.key;
if (!key) return;
if (sortKey === key) {
sortDir = -sortDir;
} else {
sortKey = key;
sortDir = key === 'uploaded_at' ? -1 : 1;
}
document.querySelectorAll('th.sortable').forEach(th => {
th.textContent = th.textContent.replace(/ [▲▼]$/, '');
});
const marker = sortDir > 0 ? ' ▲' : ' ▼';
e.target.textContent = e.target.textContent.replace(/ [▲▼]$/, '') + marker;
renderList();
}
$('#search').addEventListener('input', renderList);
$('#refresh').addEventListener('click', loadUploads);
$('#set-token').addEventListener('click', () => openTokenModal({ allowCancel: true }));
$('#logout').addEventListener('click', () => {
localStorage.removeItem('adminToken');
openTokenModal({ error: 'Token cleared. Enter a new token to continue.', allowCancel: false });
});
$('#token-form').addEventListener('submit', saveTokenFromModal);
$('#modal-cancel').addEventListener('click', closeTokenModal);
document.querySelectorAll('th.sortable').forEach(th => th.addEventListener('click', handleSort));
if (!localStorage.getItem('adminToken')) {
openTokenModal({ allowCancel: false });
} else {
loadUploads();
}
</script>
</body>
</html>
+2
View File
@@ -18,6 +18,8 @@ const (
ActionClusterCreate Action = "cluster.create" ActionClusterCreate Action = "cluster.create"
ActionClusterUpdate Action = "cluster.update" ActionClusterUpdate Action = "cluster.update"
ActionClusterDelete Action = "cluster.delete" ActionClusterDelete Action = "cluster.delete"
ActionUploadCreate Action = "upload.create"
ActionUploadDelete Action = "upload.delete"
) )
// Entry is one admin write operation recorded for accountability. // Entry is one admin write operation recorded for accountability.
+2
View File
@@ -5,6 +5,7 @@ import (
"time" "time"
"spark-mcp-go/internal/analyzer" "spark-mcp-go/internal/analyzer"
"spark-mcp-go/internal/audit"
"spark-mcp-go/internal/httpclient" "spark-mcp-go/internal/httpclient"
"spark-mcp-go/internal/storage" "spark-mcp-go/internal/storage"
"spark-mcp-go/internal/uploads" "spark-mcp-go/internal/uploads"
@@ -13,6 +14,7 @@ import (
// Deps bundles the dependencies shared by all MCP Tool handlers. // Deps bundles the dependencies shared by all MCP Tool handlers.
type Deps struct { type Deps struct {
Logger *slog.Logger Logger *slog.Logger
AuditRepo *audit.Repo
ClusterRepo *storage.ClusterRepo ClusterRepo *storage.ClusterRepo
SparkSubmitTimeout time.Duration SparkSubmitTimeout time.Duration
HTTPClient *httpclient.Client HTTPClient *httpclient.Client
+24 -7
View File
@@ -16,6 +16,7 @@ import (
"github.com/mark3labs/mcp-go/mcp" "github.com/mark3labs/mcp-go/mcp"
"spark-mcp-go/internal/audit"
"spark-mcp-go/internal/cluster" "spark-mcp-go/internal/cluster"
"spark-mcp-go/internal/httpclient" "spark-mcp-go/internal/httpclient"
"spark-mcp-go/internal/storage" "spark-mcp-go/internal/storage"
@@ -24,7 +25,7 @@ import (
// testDepsWithDataDir returns dependencies backed by an in-memory DB and a // testDepsWithDataDir returns dependencies backed by an in-memory DB and a
// temporary data directory. // temporary data directory.
func testDepsWithDataDir(t *testing.T) (*Deps, *storage.ClusterRepo) { func testDepsWithDataDir(t *testing.T) (*Deps, *storage.ClusterRepo, *audit.Repo) {
t.Helper() t.Helper()
db, err := storage.Open(":memory:") db, err := storage.Open(":memory:")
if err != nil { if err != nil {
@@ -43,11 +44,12 @@ func testDepsWithDataDir(t *testing.T) (*Deps, *storage.ClusterRepo) {
Timeout: 5 * time.Second, Timeout: 5 * time.Second,
MaxResponseBytes: 1 << 20, MaxResponseBytes: 1 << 20,
}), }),
AuditRepo: audit.NewRepo(db),
ClusterRepo: db.Clusters(), ClusterRepo: db.Clusters(),
MaxResponseBytes: 1 << 20, MaxResponseBytes: 1 << 20,
DataDir: dataDir, DataDir: dataDir,
UploadStore: uploadStore, UploadStore: uploadStore,
}, db.Clusters() }, db.Clusters(), audit.NewRepo(db)
} }
// createCluster creates a cluster in the repository with the given fields. // createCluster creates a cluster in the repository with the given fields.
@@ -98,7 +100,7 @@ func TestFetchURL(t *testing.T) {
})) }))
defer srv.Close() defer srv.Close()
deps, repo := testDepsWithDataDir(t) deps, repo, _ := testDepsWithDataDir(t)
createCluster(t, repo, &cluster.Cluster{ createCluster(t, repo, &cluster.Cluster{
ID: "cluster-a", ID: "cluster-a",
Name: "Cluster A", Name: "Cluster A",
@@ -292,7 +294,7 @@ func TestFetchURL_RedirectPreservesAuth(t *testing.T) {
})) }))
defer server1.Close() defer server1.Close()
deps, repo := testDepsWithDataDir(t) deps, repo, _ := testDepsWithDataDir(t)
createCluster(t, repo, &cluster.Cluster{ createCluster(t, repo, &cluster.Cluster{
ID: "cluster-redirect", ID: "cluster-redirect",
Name: "Redirect", Name: "Redirect",
@@ -321,7 +323,7 @@ func TestFetchURL_RedirectPreservesAuth(t *testing.T) {
} }
func TestUploadFile(t *testing.T) { func TestUploadFile(t *testing.T) {
deps, _ := testDepsWithDataDir(t) deps, _, auditRepo := testDepsWithDataDir(t)
req := newToolRequest(UploadFileName, map[string]any{ req := newToolRequest(UploadFileName, map[string]any{
"filename": "hello.txt", "filename": "hello.txt",
@@ -367,10 +369,25 @@ func TestUploadFile(t *testing.T) {
if info.Mode().Perm() != 0o640 { if info.Mode().Perm() != 0o640 {
t.Errorf("mode=%o, want %o", info.Mode().Perm(), 0o640) t.Errorf("mode=%o, want %o", info.Mode().Perm(), 0o640)
} }
entries, err := auditRepo.List(context.Background(), 100)
if err != nil {
t.Fatalf("list audit: %v", err)
}
found := false
for _, e := range entries {
if e.Action == audit.ActionUploadCreate && e.Actor == "tool:upload_file" && e.ClusterID == fileID {
found = true
break
}
}
if !found {
t.Errorf("missing upload.create audit entry for file_id %s", fileID)
}
} }
func TestUploadFile_PathTraversal(t *testing.T) { func TestUploadFile_PathTraversal(t *testing.T) {
deps, _ := testDepsWithDataDir(t) deps, _, _ := testDepsWithDataDir(t)
cases := []string{ cases := []string{
"../../../etc/passwd", "../../../etc/passwd",
@@ -401,7 +418,7 @@ func TestUploadFile_PathTraversal(t *testing.T) {
} }
func TestUploadFile_Base64(t *testing.T) { func TestUploadFile_Base64(t *testing.T) {
deps, _ := testDepsWithDataDir(t) deps, _, _ := testDepsWithDataDir(t)
req := newToolRequest(UploadFileName, map[string]any{ req := newToolRequest(UploadFileName, map[string]any{
"filename": "hello.bin", "filename": "hello.bin",
+5 -5
View File
@@ -14,7 +14,7 @@ import (
) )
func TestSparkSubmit_StructuredCommand(t *testing.T) { func TestSparkSubmit_StructuredCommand(t *testing.T) {
deps, repo := testDepsWithDataDir(t) deps, repo, _ := testDepsWithDataDir(t)
deps.SparkSubmitTimeout = 5 * time.Second deps.SparkSubmitTimeout = 5 * time.Second
store := deps.UploadStore store := deps.UploadStore
@@ -98,7 +98,7 @@ func TestSparkSubmit_StructuredCommand(t *testing.T) {
} }
func TestSparkSubmit_MissingRequiredField(t *testing.T) { func TestSparkSubmit_MissingRequiredField(t *testing.T) {
deps, _ := testDepsWithDataDir(t) deps, _, _ := testDepsWithDataDir(t)
store := deps.UploadStore store := deps.UploadStore
_, _, _, _, mintedPath, err := store.Save([]byte("# dummy\n"), "dummy.py") _, _, _, _, mintedPath, err := store.Save([]byte("# dummy\n"), "dummy.py")
if err != nil { if err != nil {
@@ -131,7 +131,7 @@ func TestSparkSubmit_MissingRequiredField(t *testing.T) {
} }
func TestSparkSubmit_BadSparkConfValue(t *testing.T) { func TestSparkSubmit_BadSparkConfValue(t *testing.T) {
deps, _ := testDepsWithDataDir(t) deps, _, _ := testDepsWithDataDir(t)
store := deps.UploadStore store := deps.UploadStore
_, _, _, _, mintedPath, err := store.Save([]byte("# dummy\n"), "dummy.py") _, _, _, _, mintedPath, err := store.Save([]byte("# dummy\n"), "dummy.py")
if err != nil { if err != nil {
@@ -166,7 +166,7 @@ func TestSparkSubmit_BadSparkConfValue(t *testing.T) {
} }
func TestSparkSubmit_RejectsNonMintedPath(t *testing.T) { func TestSparkSubmit_RejectsNonMintedPath(t *testing.T) {
deps, _ := testDepsWithDataDir(t) deps, _, _ := testDepsWithDataDir(t)
unminted := filepath.Join(t.TempDir(), "unminted.py") unminted := filepath.Join(t.TempDir(), "unminted.py")
if err := os.WriteFile(unminted, []byte("print('not from upload_file')\n"), 0o644); err != nil { if err := os.WriteFile(unminted, []byte("print('not from upload_file')\n"), 0o644); err != nil {
@@ -200,7 +200,7 @@ func TestSparkSubmit_RejectsNonMintedPath(t *testing.T) {
} }
func TestSparkSubmit_EmptyMaster(t *testing.T) { func TestSparkSubmit_EmptyMaster(t *testing.T) {
deps, _ := testDepsWithDataDir(t) deps, _, _ := testDepsWithDataDir(t)
store := deps.UploadStore store := deps.UploadStore
_, _, _, _, mintedPath, err := store.Save([]byte("# dummy\n"), "dummy.py") _, _, _, _, mintedPath, err := store.Save([]byte("# dummy\n"), "dummy.py")
if err != nil { if err != nil {
+17
View File
@@ -8,6 +8,8 @@ import (
"regexp" "regexp"
"github.com/mark3labs/mcp-go/mcp" "github.com/mark3labs/mcp-go/mcp"
"spark-mcp-go/internal/audit"
) )
const UploadFileName = "upload_file" const UploadFileName = "upload_file"
@@ -76,6 +78,21 @@ func (d *Deps) UploadFileHandler(ctx context.Context, req mcp.CallToolRequest) (
return errResult("upload_file: save upload: " + err.Error()), nil return errResult("upload_file: save upload: " + err.Error()), nil
} }
if d.AuditRepo != nil {
details, _ := audit.MarshalDetails(map[string]any{
"file_id": fileID,
"name": filename,
"size": size,
"sha256": sha256Hex,
})
_ = d.AuditRepo.Insert(ctx, &audit.Entry{
Actor: "tool:upload_file",
Action: audit.ActionUploadCreate,
ClusterID: fileID,
Details: details,
})
}
result := map[string]any{ result := map[string]any{
"file_id": fileID, "file_id": fileID,
"path": absPath, "path": absPath,
+9
View File
@@ -38,6 +38,15 @@ CREATE TABLE IF NOT EXISTS audit_log (
details TEXT details TEXT
); );
CREATE INDEX IF NOT EXISTS idx_audit_ts ON audit_log(timestamp DESC); CREATE INDEX IF NOT EXISTS idx_audit_ts ON audit_log(timestamp DESC);
CREATE TABLE IF NOT EXISTS upload_files (
file_id TEXT PRIMARY KEY,
name TEXT NOT NULL,
size INTEGER NOT NULL,
sha256 TEXT NOT NULL,
uploaded_at INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_uploads_ts ON upload_files(uploaded_at DESC);
` `
// DB is the storage handle. // DB is the storage handle.
+147
View File
@@ -0,0 +1,147 @@
package storage
import (
"context"
"database/sql"
"errors"
"fmt"
"strings"
"time"
)
// UploadMeta is the database index record for an uploaded file.
//
// The .meta.json sidecar remains the source of truth; this struct is the
// queryable mirror used by the admin UI.
type UploadMeta struct {
FileID string `json:"file_id"`
Name string `json:"name"`
Size int64 `json:"size"`
SHA256 string `json:"sha256"`
UploadedAt time.Time `json:"uploaded_at"`
}
// UploadRepo provides CRUD operations for upload file index records.
type UploadRepo struct {
db *DB
}
// Uploads returns a repository bound to this DB handle.
func (d *DB) Uploads() *UploadRepo {
return &UploadRepo{db: d}
}
// Create inserts a new upload index record. Timestamps are stored as Unix
// nanoseconds for stable ORDER BY semantics.
func (r *UploadRepo) Create(ctx context.Context, fileID, name string, size int64, sha256Hex string, uploadedAt time.Time) error {
_, err := r.db.sqlDB.ExecContext(ctx, `
INSERT INTO upload_files (file_id, name, size, sha256, uploaded_at)
VALUES (?, ?, ?, ?, ?)`,
fileID, name, size, sha256Hex, uploadedAt.UnixNano(),
)
if err != nil {
if strings.Contains(err.Error(), "UNIQUE constraint failed") {
return fmt.Errorf("storage: create upload %s: primary key conflict: %w", fileID, err)
}
return fmt.Errorf("storage: create upload %s: %w", fileID, err)
}
return nil
}
// Get retrieves an upload index record by file_id. If the record does not
// exist, it returns ErrNotFound.
func (r *UploadRepo) Get(ctx context.Context, fileID string) (UploadMeta, error) {
row := r.db.sqlDB.QueryRowContext(ctx, `
SELECT file_id, name, size, sha256, uploaded_at
FROM upload_files
WHERE file_id = ?`, fileID)
m, err := scanUpload(row.Scan)
if errors.Is(err, sql.ErrNoRows) {
return UploadMeta{}, ErrNotFound
}
if err != nil {
return UploadMeta{}, fmt.Errorf("storage: get upload %s: %w", fileID, err)
}
return m, nil
}
// List returns upload index records ordered by uploaded_at descending.
//
// If search is non-empty, name is filtered with a case-insensitive LIKE.
// A limit of zero or less falls back to 100; values above 500 are capped.
func (r *UploadRepo) List(ctx context.Context, search string, limit int) ([]UploadMeta, error) {
if limit <= 0 {
limit = 100
}
if limit > 500 {
limit = 500
}
var rows *sql.Rows
var err error
if search != "" {
rows, err = r.db.sqlDB.QueryContext(ctx, `
SELECT file_id, name, size, sha256, uploaded_at
FROM upload_files
WHERE name LIKE ?
ORDER BY uploaded_at DESC
LIMIT ?`,
"%"+search+"%", limit,
)
} else {
rows, err = r.db.sqlDB.QueryContext(ctx, `
SELECT file_id, name, size, sha256, uploaded_at
FROM upload_files
ORDER BY uploaded_at DESC
LIMIT ?`, limit)
}
if err != nil {
return nil, fmt.Errorf("storage: list uploads: %w", err)
}
defer rows.Close()
var out []UploadMeta
for rows.Next() {
m, err := scanUpload(rows.Scan)
if err != nil {
return nil, fmt.Errorf("storage: list uploads: %w", err)
}
out = append(out, m)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("storage: list uploads: %w", err)
}
if out == nil {
// Force empty slice (not null) in JSON.
out = []UploadMeta{}
}
return out, nil
}
// Delete removes an upload index record by file_id. If the record does not
// exist, it returns ErrNotFound.
func (r *UploadRepo) Delete(ctx context.Context, fileID string) error {
res, err := r.db.sqlDB.ExecContext(ctx, `DELETE FROM upload_files WHERE file_id = ?`, fileID)
if err != nil {
return fmt.Errorf("storage: delete upload %s: %w", fileID, err)
}
n, err := res.RowsAffected()
if err != nil {
return fmt.Errorf("storage: delete upload %s: rows affected: %w", fileID, err)
}
if n == 0 {
return ErrNotFound
}
return nil
}
func scanUpload(scan scanFunc) (UploadMeta, error) {
var m UploadMeta
var uploadedAt int64
if err := scan(&m.FileID, &m.Name, &m.Size, &m.SHA256, &uploadedAt); err != nil {
return UploadMeta{}, err
}
m.UploadedAt = time.Unix(0, uploadedAt)
return m, nil
}
+162
View File
@@ -0,0 +1,162 @@
package storage
import (
"context"
"errors"
"fmt"
"testing"
"time"
)
func newUploadRepo(t *testing.T) *UploadRepo {
t.Helper()
db, err := Open(":memory:")
if err != nil {
t.Fatalf("Open in-memory db: %v", err)
}
t.Cleanup(func() { _ = db.Close() })
return db.Uploads()
}
func assertUploadEqual(t *testing.T, got, want UploadMeta) {
t.Helper()
if got.FileID != want.FileID {
t.Errorf("FileID: got %q, want %q", got.FileID, want.FileID)
}
if got.Name != want.Name {
t.Errorf("Name: got %q, want %q", got.Name, want.Name)
}
if got.Size != want.Size {
t.Errorf("Size: got %d, want %d", got.Size, want.Size)
}
if got.SHA256 != want.SHA256 {
t.Errorf("SHA256: got %q, want %q", got.SHA256, want.SHA256)
}
if !got.UploadedAt.Equal(want.UploadedAt) {
t.Errorf("UploadedAt: got %v, want %v", got.UploadedAt, want.UploadedAt)
}
}
func TestUploadRepo_CreateGetListDelete(t *testing.T) {
repo := newUploadRepo(t)
ctx := context.Background()
uploadedAt := time.Unix(0, time.Now().UnixNano())
want := UploadMeta{
FileID: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
Name: "report.csv",
Size: 42,
SHA256: "deadbeef",
UploadedAt: uploadedAt,
}
if err := repo.Create(ctx, want.FileID, want.Name, want.Size, want.SHA256, want.UploadedAt); err != nil {
t.Fatalf("Create: %v", err)
}
got, err := repo.Get(ctx, want.FileID)
if err != nil {
t.Fatalf("Get: %v", err)
}
assertUploadEqual(t, got, want)
if err := repo.Create(ctx, want.FileID, "other", 1, "abcd", time.Now()); err == nil {
t.Errorf("duplicate Create succeeded, want error")
}
list, err := repo.List(ctx, "", 100)
if err != nil {
t.Fatalf("List: %v", err)
}
if len(list) != 1 {
t.Fatalf("List len: got %d, want 1", len(list))
}
assertUploadEqual(t, list[0], want)
if err := repo.Delete(ctx, want.FileID); err != nil {
t.Fatalf("Delete: %v", err)
}
if _, err := repo.Get(ctx, want.FileID); !errors.Is(err, ErrNotFound) {
t.Fatalf("Get after Delete: got %v, want ErrNotFound", err)
}
if err := repo.Delete(ctx, want.FileID); !errors.Is(err, ErrNotFound) {
t.Fatalf("Delete again: got %v, want ErrNotFound", err)
}
}
func TestUploadRepo_ListOrderAndSearch(t *testing.T) {
repo := newUploadRepo(t)
ctx := context.Background()
base := time.Unix(0, time.Now().UnixNano())
records := []UploadMeta{
{FileID: "00000000000000000000000000000001", Name: "alpha.csv", Size: 1, SHA256: "a", UploadedAt: base.Add(-2 * time.Hour)},
{FileID: "00000000000000000000000000000002", Name: "beta.log", Size: 2, SHA256: "b", UploadedAt: base.Add(-1 * time.Hour)},
{FileID: "00000000000000000000000000000003", Name: "gamma.csv", Size: 3, SHA256: "c", UploadedAt: base},
}
for _, r := range records {
if err := repo.Create(ctx, r.FileID, r.Name, r.Size, r.SHA256, r.UploadedAt); err != nil {
t.Fatalf("Create %s: %v", r.FileID, err)
}
}
all, err := repo.List(ctx, "", 100)
if err != nil {
t.Fatalf("List: %v", err)
}
if len(all) != 3 {
t.Fatalf("List len: got %d, want 3", len(all))
}
wantOrder := []string{"00000000000000000000000000000003", "00000000000000000000000000000002", "00000000000000000000000000000001"}
for i, id := range wantOrder {
if all[i].FileID != id {
t.Errorf("List order[%d]: got %q, want %q", i, all[i].FileID, id)
}
}
csv, err := repo.List(ctx, "csv", 100)
if err != nil {
t.Fatalf("List search: %v", err)
}
if len(csv) != 2 {
t.Fatalf("search csv len: got %d, want 2", len(csv))
}
if csv[0].Name != "gamma.csv" || csv[1].Name != "alpha.csv" {
t.Errorf("search csv order: got %v", []string{csv[0].Name, csv[1].Name})
}
beta, err := repo.List(ctx, "beta", 100)
if err != nil {
t.Fatalf("List search beta: %v", err)
}
if len(beta) != 1 || beta[0].Name != "beta.log" {
t.Errorf("search beta: got %+v", beta)
}
}
func TestUploadRepo_ListLimitCap(t *testing.T) {
repo := newUploadRepo(t)
ctx := context.Background()
base := time.Unix(0, time.Now().UnixNano())
for i := 0; i < 10; i++ {
id := fmt.Sprintf("%032d", i)
if err := repo.Create(ctx, id, "x", int64(i), "h", base.Add(time.Duration(i)*time.Second)); err != nil {
t.Fatalf("Create %d: %v", i, err)
}
}
list, err := repo.List(ctx, "", 0)
if err != nil {
t.Fatalf("List default limit: %v", err)
}
// Default limit is 100 but only 10 rows exist.
if len(list) != 10 {
t.Errorf("default limit: got %d, want 10", len(list))
}
list, err = repo.List(ctx, "", 501)
if err != nil {
t.Fatalf("List cap: %v", err)
}
if len(list) != 10 {
t.Errorf("cap limit: got %d, want 10", len(list))
}
// Explicit small limit is respected.
list, err = repo.List(ctx, "", 3)
if err != nil {
t.Fatalf("List small limit: %v", err)
}
if len(list) != 3 {
t.Errorf("small limit: got %d, want 3", len(list))
}
}
+86 -5
View File
@@ -9,12 +9,16 @@ import (
"crypto/sha256" "crypto/sha256"
"encoding/hex" "encoding/hex"
"encoding/json" "encoding/json"
"errors"
"fmt" "fmt"
"log/slog"
"os" "os"
"path/filepath" "path/filepath"
"regexp" "regexp"
"strings" "strings"
"time" "time"
"spark-mcp-go/internal/storage"
) )
var fileIDRegex = regexp.MustCompile(`^[0-9a-f]{32}$`) var fileIDRegex = regexp.MustCompile(`^[0-9a-f]{32}$`)
@@ -22,6 +26,20 @@ var fileIDRegex = regexp.MustCompile(`^[0-9a-f]{32}$`)
// Store is a local file store rooted at Root. // Store is a local file store rooted at Root.
type Store struct { type Store struct {
Root string Root string
repo UploadsDB
}
// UploadsDB is the minimal surface the Store needs from an upload index.
type UploadsDB interface {
Create(ctx context.Context, fileID, name string, size int64, sha256Hex string, uploadedAt time.Time) error
Get(ctx context.Context, fileID string) (storage.UploadMeta, error)
}
// SetRepo binds an upload index repository for dual-write. The DB is an
// index only; failures are logged but never fail the file write because
// .meta.json is the source of truth and startup backfill can recover.
func (s *Store) SetRepo(repo UploadsDB) {
s.repo = repo
} }
// New creates a Store rooted at root. It creates root with mode 0o750 if it // New creates a Store rooted at root. It creates root with mode 0o750 if it
@@ -111,6 +129,13 @@ func (s *Store) Save(data []byte, originalName string) (fileID, name string, siz
return "", "", 0, "", "", fmt.Errorf("uploads: write sidecar: %w", err) return "", "", 0, "", "", fmt.Errorf("uploads: write sidecar: %w", err)
} }
// Index the upload in the DB as a best-effort mirror of the sidecar.
if s.repo != nil {
if dbErr := s.repo.Create(context.Background(), fileID, originalName, meta.Size, sha256Hex, meta.UploadedAt); dbErr != nil {
slog.Default().Warn("uploads: failed to index upload in DB", "file_id", fileID, "err", dbErr)
}
}
return fileID, originalName, meta.Size, sha256Hex, absPath, nil return fileID, originalName, meta.Size, sha256Hex, absPath, nil
} }
@@ -240,6 +265,54 @@ func (s *Store) Sweep(ctx context.Context, ttl time.Duration) (deleted int, err
return deleted, nil return deleted, nil
} }
// Backfill walks Root and inserts a DB index row for every existing
// .meta.json sidecar that is not already indexed. It is the recovery path
// for sidecars created before the DB table existed.
func (s *Store) Backfill(ctx context.Context, repo UploadsDB) (inserted int, err error) {
entries, err := os.ReadDir(s.Root)
if err != nil {
return 0, fmt.Errorf("uploads: backfill read root: %w", err)
}
for _, e := range entries {
if err := ctx.Err(); err != nil {
return inserted, err
}
name := e.Name()
base := strings.TrimSuffix(name, ".meta.json")
if base == name || !fileIDRegex.MatchString(base) {
continue
}
metaPath := filepath.Join(s.Root, name)
sc, err := readSidecar(metaPath)
if err != nil {
// Skip corrupt sidecars; Sweep will reap them later.
continue
}
_, getErr := repo.Get(ctx, base)
if getErr == nil {
continue
}
if !errors.Is(getErr, storage.ErrNotFound) {
return inserted, fmt.Errorf("uploads: backfill lookup %s: %w", base, getErr)
}
createErr := repo.Create(ctx, base, sc.Name, sc.Size, sc.Sha256, sc.UploadedAt)
if createErr != nil {
if strings.Contains(createErr.Error(), "UNIQUE constraint failed") {
continue
}
return inserted, fmt.Errorf("uploads: backfill index %s: %w", base, createErr)
}
inserted++
}
return inserted, nil
}
func newFileID() (string, error) { func newFileID() (string, error) {
var b [16]byte var b [16]byte
if _, err := rand.Read(b[:]); err != nil { if _, err := rand.Read(b[:]); err != nil {
@@ -249,16 +322,24 @@ func newFileID() (string, error) {
} }
func readUploadedAt(path string) (time.Time, error) { func readUploadedAt(path string) (time.Time, error) {
data, err := os.ReadFile(path) sc, err := readSidecar(path)
if err != nil { if err != nil {
return time.Time{}, err return time.Time{}, err
} }
var sc sidecar
if err := json.Unmarshal(data, &sc); err != nil {
return time.Time{}, err
}
if sc.UploadedAt.IsZero() { if sc.UploadedAt.IsZero() {
return time.Time{}, fmt.Errorf("uploads: missing uploaded_at") return time.Time{}, fmt.Errorf("uploads: missing uploaded_at")
} }
return sc.UploadedAt, nil return sc.UploadedAt, nil
} }
func readSidecar(path string) (sidecar, error) {
var sc sidecar
data, err := os.ReadFile(path)
if err != nil {
return sc, err
}
if err := json.Unmarshal(data, &sc); err != nil {
return sc, err
}
return sc, nil
}
+180
View File
@@ -5,12 +5,15 @@ import (
"crypto/sha256" "crypto/sha256"
"encoding/hex" "encoding/hex"
"encoding/json" "encoding/json"
"errors"
"fmt" "fmt"
"os" "os"
"path/filepath" "path/filepath"
"regexp" "regexp"
"testing" "testing"
"time" "time"
"spark-mcp-go/internal/storage"
) )
func TestStore_SaveAndRetrieve(t *testing.T) { func TestStore_SaveAndRetrieve(t *testing.T) {
@@ -351,3 +354,180 @@ func TestStore_Sweep_DeletesDataWithCorruptSidecar(t *testing.T) {
t.Errorf("corrupt sidecar still exists") t.Errorf("corrupt sidecar still exists")
} }
} }
type fakeUploadRepo struct {
calls []storage.UploadMeta
err error
}
func (f *fakeUploadRepo) Create(ctx context.Context, fileID, name string, size int64, sha256Hex string, uploadedAt time.Time) error {
f.calls = append(f.calls, storage.UploadMeta{
FileID: fileID, Name: name, Size: size, SHA256: sha256Hex, UploadedAt: uploadedAt,
})
return f.err
}
func (f *fakeUploadRepo) Get(ctx context.Context, fileID string) (storage.UploadMeta, error) {
for _, m := range f.calls {
if m.FileID == fileID {
return m, nil
}
}
return storage.UploadMeta{}, storage.ErrNotFound
}
func (f *fakeUploadRepo) List(ctx context.Context, search string, limit int) ([]storage.UploadMeta, error) {
return f.calls, nil
}
func (f *fakeUploadRepo) Delete(ctx context.Context, fileID string) error {
return nil
}
func TestStore_Save_DualWrite(t *testing.T) {
store, err := New(t.TempDir())
if err != nil {
t.Fatalf("new store: %v", err)
}
repo := &fakeUploadRepo{}
store.SetRepo(repo)
data := []byte("dual-write test")
fileID, _, size, sha256Hex, absPath, err := store.Save(data, "dual.txt")
if err != nil {
t.Fatalf("save: %v", err)
}
if absPath == "" {
t.Fatal("absPath empty")
}
if len(repo.calls) != 1 {
t.Fatalf("repo.Create calls: got %d, want 1", len(repo.calls))
}
call := repo.calls[0]
if call.FileID != fileID {
t.Errorf("FileID: got %q, want %q", call.FileID, fileID)
}
if call.Name != "dual.txt" {
t.Errorf("Name: got %q, want dual.txt", call.Name)
}
if call.Size != size {
t.Errorf("Size: got %d, want %d", call.Size, size)
}
if call.SHA256 != sha256Hex {
t.Errorf("SHA256: got %q, want %q", call.SHA256, sha256Hex)
}
if call.UploadedAt.IsZero() {
t.Errorf("UploadedAt is zero")
}
}
func TestStore_Save_RepoErrorDoesNotFailUpload(t *testing.T) {
store, err := New(t.TempDir())
if err != nil {
t.Fatalf("new store: %v", err)
}
repo := &fakeUploadRepo{err: errors.New("db down")}
store.SetRepo(repo)
data := []byte("db error test")
fileID, _, _, _, absPath, err := store.Save(data, "error.txt")
if err != nil {
t.Fatalf("save should not fail when repo errors: %v", err)
}
if _, err := os.Stat(absPath); err != nil {
t.Errorf("data file missing: %v", err)
}
if _, err := os.Stat(absPath + ".meta.json"); err != nil {
t.Errorf("sidecar missing: %v", err)
}
if len(repo.calls) != 1 {
t.Errorf("repo.Create calls: got %d, want 1", len(repo.calls))
}
if repo.calls[0].FileID != fileID {
t.Errorf("repo call file_id: got %q, want %q", repo.calls[0].FileID, fileID)
}
}
func TestStore_Backfill(t *testing.T) {
store, err := New(t.TempDir())
if err != nil {
t.Fatalf("new store: %v", err)
}
repo := &fakeUploadRepo{}
now := time.Now()
ids := []string{
"00000000000000000000000000000001",
"00000000000000000000000000000002",
"00000000000000000000000000000003",
}
for i, id := range ids {
sc := map[string]any{
"name": fmt.Sprintf("file%d.txt", i),
"size": i + 1,
"sha256": fmt.Sprintf("sha%d", i),
"uploaded_at": now.Add(time.Duration(i) * time.Second).Format(time.RFC3339Nano),
}
b, _ := json.Marshal(sc)
if err := os.WriteFile(filepath.Join(store.Root, id+".meta.json"), b, 0o600); err != nil {
t.Fatalf("create sidecar %s: %v", id, err)
}
}
inserted, err := store.Backfill(context.Background(), repo)
if err != nil {
t.Fatalf("backfill: %v", err)
}
if inserted != 3 {
t.Errorf("inserted=%d, want 3", inserted)
}
if len(repo.calls) != 3 {
t.Errorf("repo.Create calls: got %d, want 3", len(repo.calls))
}
// Second backfill must skip existing rows.
inserted, err = store.Backfill(context.Background(), repo)
if err != nil {
t.Fatalf("second backfill: %v", err)
}
if inserted != 0 {
t.Errorf("second inserted=%d, want 0", inserted)
}
if len(repo.calls) != 3 {
t.Errorf("repo.Create calls after second backfill: got %d, want 3", len(repo.calls))
}
}
func TestStore_Backfill_SkipsExisting(t *testing.T) {
store, err := New(t.TempDir())
if err != nil {
t.Fatalf("new store: %v", err)
}
repo := &fakeUploadRepo{}
id := "00000000000000000000000000000004"
now := time.Now()
sc := map[string]any{
"name": "existing.txt",
"size": 5,
"sha256": "sha",
"uploaded_at": now.Format(time.RFC3339Nano),
}
b, _ := json.Marshal(sc)
if err := os.WriteFile(filepath.Join(store.Root, id+".meta.json"), b, 0o600); err != nil {
t.Fatalf("create sidecar: %v", err)
}
// Pre-seed the repo so the sidecar is already indexed.
if err := repo.Create(context.Background(), id, "existing.txt", 5, "sha", now); err != nil {
t.Fatalf("seed repo: %v", err)
}
inserted, err := store.Backfill(context.Background(), repo)
if err != nil {
t.Fatalf("backfill: %v", err)
}
if inserted != 0 {
t.Errorf("inserted=%d, want 0", inserted)
}
}
+15 -1
View File
@@ -84,6 +84,19 @@ func run() error {
logger.Info("uploads.sweep", "deleted", n) logger.Info("uploads.sweep", "deleted", n)
} }
backfillCtx, backfillCancel := context.WithTimeout(context.Background(), startupSweepTimeout)
n, err = uploadStore.Backfill(backfillCtx, db.Uploads())
backfillCancel()
if errors.Is(err, context.DeadlineExceeded) {
logger.Warn("uploads.backfill.timeout", "inserted", n, "err", err)
} else if err != nil {
logger.Error("uploads.backfill", "err", err)
} else {
logger.Info("uploads.backfill", "inserted", n)
}
uploadStore.SetRepo(db.Uploads())
gin.SetMode(cfg.GinMode) gin.SetMode(cfg.GinMode)
r := gin.New() r := gin.New()
r.Use(gin.Recovery()) r.Use(gin.Recovery())
@@ -92,10 +105,11 @@ func run() error {
c.JSON(http.StatusOK, gin.H{"ok": true, "version": "0.0.0"}) c.JSON(http.StatusOK, gin.H{"ok": true, "version": "0.0.0"})
}) })
admin.Mount(r, db.Clusters(), audit.NewRepo(db), cfg.AdminTokens) admin.Mount(r, db.Clusters(), db.Uploads(), audit.NewRepo(db), &uploadStore, cfg.AdminTokens)
mcpHandler, err := mcpsrv.Handler(&tools.Deps{ mcpHandler, err := mcpsrv.Handler(&tools.Deps{
Logger: logger.With("component", "mcp"), Logger: logger.With("component", "mcp"),
AuditRepo: audit.NewRepo(db),
ClusterRepo: db.Clusters(), ClusterRepo: db.Clusters(),
SparkSubmitTimeout: cfg.SparkSubmitTimeout, SparkSubmitTimeout: cfg.SparkSubmitTimeout,
HTTPClient: httpclient.New(httpclient.Config{ HTTPClient: httpclient.New(httpclient.Config{