Compare commits

..

11 Commits

Author SHA1 Message Date
36d1c1512a Node autocleanup
All checks were successful
release-tag / release-image (push) Successful in 1m32s
2025-09-29 21:19:07 +02:00
aaa95410e4 Template Anpassung
All checks were successful
release-tag / release-image (push) Successful in 1m26s
2025-09-28 00:00:09 +02:00
0bf22b6120 fix self
All checks were successful
release-tag / release-image (push) Successful in 1m25s
2025-09-27 23:53:17 +02:00
b142a0b1a5 layout max to 1200
All checks were successful
release-tag / release-image (push) Successful in 1m33s
2025-09-27 23:42:35 +02:00
baedff9e9d Umstellung auf uuid
All checks were successful
release-tag / release-image (push) Successful in 1m32s
2025-09-27 21:42:32 +02:00
92e222f648 buxfix-2
All checks were successful
release-tag / release-image (push) Successful in 1m26s
2025-09-27 17:52:10 +02:00
43f1d01a8a diverse bugfixes
All checks were successful
release-tag / release-image (push) Successful in 1m32s
2025-09-27 17:39:39 +02:00
d125e3dd54 findnodes
All checks were successful
release-tag / release-image (push) Successful in 1m28s
2025-09-27 16:25:13 +02:00
6a0dc578ea fix
All checks were successful
release-tag / release-image (push) Successful in 1m27s
2025-09-27 16:00:53 +02:00
8c05ad6ffe update auf public download
All checks were successful
release-tag / release-image (push) Successful in 1m40s
2025-09-27 15:55:16 +02:00
7d0f4befe1 update
All checks were successful
release-tag / release-image (push) Successful in 1m28s
2025-09-27 14:16:15 +02:00
9 changed files with 540 additions and 108 deletions

View File

@@ -22,6 +22,14 @@ import (
"git.send.nrw/sendnrw/decent-webui/internal/mesh"
)
func parseDuration(s string, def time.Duration) time.Duration {
d, err := time.ParseDuration(strings.TrimSpace(s))
if err != nil || d <= 0 {
return def
}
return d
}
/*** Config ***/
func loadConfig() AppConfig {
// HTTP
@@ -44,6 +52,9 @@ func loadConfig() AppConfig {
DiscoveryAddress: getenvDefault("MESH_DISCOVERY_ADDR", "239.8.8.8:9898"),
}
m.PeerTTL = parseDuration(os.Getenv("MESH_PEER_TTL"), 2*time.Minute)
m.PruneInterval = parseDuration(os.Getenv("MESH_PRUNE_INTERVAL"), 30*time.Second)
// Wenn keine AdvertURL gesetzt ist, versuche eine sinnvolle Herleitung:
if strings.TrimSpace(m.AdvertURL) == "" {
m.AdvertURL = inferAdvertURL(m.BindAddr)
@@ -62,11 +73,13 @@ func loadConfig() AppConfig {
}
return AppConfig{
HTTPAddr: httpAddr,
APIKey: apiKey,
AdminUser: adminUser,
AdminPass: adminPass,
Mesh: m,
HTTPAddr: httpAddr,
APIKey: apiKey,
AdminUser: adminUser,
AdminPass: adminPass,
Mesh: m,
PublicDownloads: parseBoolEnv("PUBLIC_DOWNLOADS", false),
PublicPath: getenvDefault("PUBLIC_DOWNLOAD_PATH", "/dl"),
}
}
@@ -137,6 +150,9 @@ type AppConfig struct {
AdminUser string
AdminPass string
Mesh mesh.Config
PublicDownloads bool // ENV: PUBLIC_DOWNLOADS (default false)
PublicPath string // ENV: PUBLIC_DOWNLOAD_PATH (default "/dl")
}
/*** Middleware ***/
@@ -188,7 +204,7 @@ func writeJSON(w http.ResponseWriter, code int, v any) {
/*** API-Routen ***/
func fileRoutes(mux *http.ServeMux, store filesvc.MeshStore) {
func fileRoutes(mux *http.ServeMux, store filesvc.MeshStore, blobs blobfs.Store) {
// Health
mux.HandleFunc("/api/v1/health", func(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
@@ -198,13 +214,8 @@ func fileRoutes(mux *http.ServeMux, store filesvc.MeshStore) {
mux.HandleFunc("/api/v1/items", func(w http.ResponseWriter, r *http.Request) {
switch r.Method {
case http.MethodGet:
nextStr := r.URL.Query().Get("next")
var next filesvc.ID
if nextStr != "" {
if n, err := strconv.ParseInt(nextStr, 10, 64); err == nil {
next = filesvc.ID(n)
}
}
nextStr := strings.TrimSpace(r.URL.Query().Get("next"))
next := filesvc.ID(nextStr)
items, nextOut, err := store.List(r.Context(), next, 100)
if err != nil {
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()})
@@ -270,6 +281,7 @@ func fileRoutes(mux *http.ServeMux, store filesvc.MeshStore) {
return
}
it, err := store.Delete(r.Context(), in.ID)
_ = blobs.Delete(r.Context(), string(in.ID))
if err != nil {
status := http.StatusBadRequest
if errors.Is(err, filesvc.ErrNotFound) {
@@ -310,7 +322,7 @@ func apiFiles(mux *http.ServeMux, store filesvc.MeshStore, blobs blobfs.Store, m
writeJSON(w, http.StatusBadRequest, map[string]string{"error": err.Error()})
return
}
meta, err := blobs.Save(r.Context(), int64(it.ID), name, fh)
meta, err := blobs.Save(r.Context(), string(it.ID), name, fh)
if err != nil {
_, _ = store.Delete(r.Context(), it.ID)
writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()})
@@ -325,27 +337,51 @@ func apiFiles(mux *http.ServeMux, store filesvc.MeshStore, blobs blobfs.Store, m
// Download
mux.HandleFunc("/api/v1/files/", func(w http.ResponseWriter, r *http.Request) {
// /api/v1/files/{id}/download
parts := strings.Split(strings.TrimPrefix(r.URL.Path, "/api/v1/files/"), "/")
if len(parts) != 2 || parts[1] != "download" {
http.NotFound(w, r)
return
}
id, err := strconv.ParseInt(parts[0], 10, 64)
id := parts[0]
if strings.TrimSpace(id) == "" {
http.NotFound(w, r)
return
}
// 1) Metadaten prüfen
it, err := store.Get(r.Context(), filesvc.ID(id))
if err != nil || it.Deleted {
http.NotFound(w, r)
return
}
// 2) lokal
if rc, meta, err := blobs.Open(r.Context(), id); err == nil {
defer rc.Close()
serveBlob(w, r, rc, meta, it.Name)
return
}
// 3) remote holen & cachen
rrc, _, _, _, err := meshNode.FetchBlobAny(r.Context(), id)
if err != nil {
http.NotFound(w, r)
return
}
rc, meta, err := blobs.Open(r.Context(), id)
if err != nil {
http.NotFound(w, r)
defer rrc.Close()
if _, err := blobs.Save(r.Context(), id, it.Name, rrc); err != nil {
http.Error(w, "cache failed", http.StatusInternalServerError)
return
}
defer rc.Close()
w.Header().Set("Content-Type", meta.ContentType)
w.Header().Set("Content-Length", strconv.FormatInt(meta.Size, 10))
w.Header().Set("Content-Disposition", fmt.Sprintf(`attachment; filename="%s"`, meta.Name))
_, _ = io.Copy(w, rc)
// 4) lokal streamen
lrc, meta, err := blobs.Open(r.Context(), id)
if err != nil {
http.Error(w, "open failed", http.StatusInternalServerError)
return
}
defer lrc.Close()
serveBlob(w, r, lrc, meta, it.Name)
})
}
@@ -355,7 +391,7 @@ func toMeshSnapshot(s filesvc.Snapshot) mesh.Snapshot {
out := mesh.Snapshot{Items: make([]mesh.Item, 0, len(s.Items))}
for _, it := range s.Items {
out.Items = append(out.Items, mesh.Item{
ID: int64(it.ID),
ID: it.ID,
Name: it.Name,
UpdatedAt: it.UpdatedAt,
Deleted: it.Deleted,
@@ -368,7 +404,7 @@ func fromMeshSnapshot(ms mesh.Snapshot) filesvc.Snapshot {
out := filesvc.Snapshot{Items: make([]filesvc.File, 0, len(ms.Items))}
for _, it := range ms.Items {
out.Items = append(out.Items, filesvc.File{
ID: filesvc.ID(it.ID),
ID: it.ID,
Name: it.Name,
UpdatedAt: it.UpdatedAt,
Deleted: it.Deleted,
@@ -386,8 +422,10 @@ func main() {
st := filesvc.NewMemStore()
// Mesh starten
mcfg := mesh.FromEnv()
mnode, err := mesh.New(mcfg, mesh.Callbacks{
//mcfg := mesh.FromEnv()
blobs := blobfs.New(getenvDefault("DATA_DIR", "./data"))
mnode, err := mesh.New(cfg.Mesh, mesh.Callbacks{
GetSnapshot: func(ctx context.Context) (mesh.Snapshot, error) {
s, err := st.Snapshot(ctx)
if err != nil {
@@ -398,13 +436,28 @@ func main() {
ApplyRemote: func(ctx context.Context, s mesh.Snapshot) error {
return st.ApplyRemote(ctx, fromMeshSnapshot(s))
},
BlobOpen: func(ctx context.Context, id string) (io.ReadCloser, string, string, int64, error) { //5588
it, err := st.Get(ctx, filesvc.ID(id))
if err != nil || it.Deleted {
return nil, "", "", 0, fmt.Errorf("not found")
}
rc, meta, err := blobs.Open(ctx, id)
if err != nil {
return nil, "", "", 0, err
}
return rc, it.Name, meta.ContentType, meta.Size, nil
},
})
if err != nil {
log.Fatalf("mesh init: %v", err)
}
// Hintergrund-Pruner starten
mnode.StartPeerPruner()
go func() {
log.Printf("[mesh] listening on %s advertise %s seeds=%v discovery=%v",
mcfg.BindAddr, mcfg.AdvertURL, mcfg.Seeds, mcfg.EnableDiscovery)
cfg.Mesh.BindAddr, cfg.Mesh.AdvertURL, cfg.Mesh.Seeds, cfg.Mesh.EnableDiscovery)
if err := mnode.Serve(); err != nil && !errors.Is(err, http.ErrServerClosed) {
log.Fatalf("mesh serve: %v", err)
}
@@ -414,12 +467,16 @@ func main() {
root := http.NewServeMux()
// API (Bearer-Auth)
blobs := blobfs.New(getenvDefault("DATA_DIR", "./data"))
//blobs := blobfs.New(getenvDefault("DATA_DIR", "./data"))
apiMux := http.NewServeMux()
fileRoutes(apiMux, st)
fileRoutes(apiMux, st, blobs)
apiFiles(apiMux, st, blobs, mnode)
root.Handle("/api/", authMiddleware(cfg.APIKey, apiMux))
if cfg.PublicDownloads {
registerPublicDownloads(root, st, blobs, mnode, cfg.PublicPath)
}
// Admin-UI (optional BasicAuth via ADMIN_USER/ADMIN_PASS)
adminRoot := http.NewServeMux()
admin.Register(adminRoot, admin.Deps{Store: st, Mesh: mnode, Blob: blobs})
@@ -450,7 +507,8 @@ func main() {
// Graceful shutdown
go func() {
log.Printf("http listening on %s (api=/api/v1, admin=/admin)", cfg.Mesh.BindAddr)
log.Printf("http listening on %s (api=/api/v1, admin=/admin)", cfg.HTTPAddr)
log.Printf("mesh listening on %s", cfg.Mesh.BindAddr)
if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
log.Fatalf("http server: %v", err)
}
@@ -467,3 +525,83 @@ func main() {
_ = mnode.Close(ctx)
log.Println("shutdown complete")
}
// Public: GET {base}/{id}
// Beispiel: /dl/1
func registerPublicDownloads(mux *http.ServeMux, store filesvc.MeshStore, blobs blobfs.Store, meshNode *mesh.Node, base string) {
if !strings.HasPrefix(base, "/") {
base = "/" + base
}
mux.HandleFunc(base+"/", func(w http.ResponseWriter, r *http.Request) {
idStr := strings.TrimPrefix(r.URL.Path, base+"/")
if idStr == "" {
http.NotFound(w, r)
return
}
id := idStr
if strings.TrimSpace(id) == "" {
http.NotFound(w, r)
return
}
// 1) Metadaten prüfen (nicht vorhanden oder gelöscht → 404)
it, err := store.Get(r.Context(), filesvc.ID(id))
if err != nil || it.Deleted {
http.NotFound(w, r)
return
}
// 2) lokal versuchen
if rc, meta, err := blobs.Open(r.Context(), id); err == nil {
defer rc.Close()
serveBlob(w, r, rc, meta, it.Name) // Download-Name aus Store!
return
}
// 3) aus Mesh holen (signiert) und cachen
rrc, _, _, _, err := meshNode.FetchBlobAny(r.Context(), id)
if err != nil {
http.NotFound(w, r)
return
}
defer rrc.Close()
if _, err := blobs.Save(r.Context(), id, it.Name, rrc); err != nil {
http.Error(w, "cache failed", http.StatusInternalServerError)
return
}
// 4) erneut lokal öffnen und streamen
lrc, meta, err := blobs.Open(r.Context(), id)
if err != nil {
http.Error(w, "open failed", http.StatusInternalServerError)
return
}
defer lrc.Close()
serveBlob(w, r, lrc, meta, it.Name)
})
}
// Hilfsfunktion: setzt sinnvolle Header und streamt die Datei
func serveBlob(w http.ResponseWriter, r *http.Request, rc io.ReadSeeker, meta blobfs.Meta, downloadName string) {
if meta.SHA256 != "" {
etag := `W/"` + meta.SHA256 + `"`
if r.Header.Get("If-None-Match") == etag {
w.WriteHeader(http.StatusNotModified)
return
}
w.Header().Set("ETag", etag)
}
if meta.ContentType == "" {
meta.ContentType = "application/octet-stream"
}
if downloadName == "" {
downloadName = meta.Name
}
w.Header().Set("Content-Type", meta.ContentType)
w.Header().Set("Content-Length", strconv.FormatInt(meta.Size, 10))
w.Header().Set("Content-Disposition", fmt.Sprintf(`attachment; filename="%s"`, downloadName))
w.Header().Set("Access-Control-Expose-Headers", "Content-Disposition")
w.Header().Set("X-Robots-Tag", "noindex")
_, _ = io.Copy(w, rc)
}

View File

@@ -56,17 +56,11 @@ func Register(mux *http.ServeMux, d Deps) {
// Partials
mux.HandleFunc("/admin/items", func(w http.ResponseWriter, r *http.Request) {
nextQ := strings.TrimSpace(r.URL.Query().Get("next"))
var nextID filesvc.ID
if nextQ != "" {
if n, err := strconv.ParseInt(nextQ, 10, 64); err == nil {
nextID = filesvc.ID(n)
}
}
nextID := filesvc.ID(strings.TrimSpace(r.URL.Query().Get("next")))
items, nextOut, _ := d.Store.List(r.Context(), nextID, 100)
type row struct {
ID int64
ID string
Name string
UpdatedAt int64
HasBlob bool
@@ -74,9 +68,9 @@ func Register(mux *http.ServeMux, d Deps) {
}
rows := make([]row, 0, len(items))
for _, it := range items {
meta, ok, _ := d.Blob.Stat(r.Context(), int64(it.ID))
meta, ok, _ := d.Blob.Stat(r.Context(), it.ID)
rows = append(rows, row{
ID: int64(it.ID),
ID: it.ID,
Name: it.Name,
UpdatedAt: it.UpdatedAt,
HasBlob: ok,
@@ -119,14 +113,14 @@ func Register(mux *http.ServeMux, d Deps) {
}
// 2) Blob speichern
if _, err := d.Blob.Save(r.Context(), int64(it.ID), name, fh); err != nil {
if _, err := d.Blob.Save(r.Context(), (it.ID), name, fh); err != nil {
// zurückrollen (Tombstone)
_, _ = d.Store.Delete(r.Context(), it.ID)
http.Error(w, "save failed: "+err.Error(), http.StatusInternalServerError)
return
}
_ = d.Mesh.SyncNow(r.Context()) // best-effort Push
http.Redirect(w, r, "/admin/items", http.StatusSeeOther)
http.Redirect(w, r, "/admin", http.StatusSeeOther)
})
// Download (Admin BasicAuth schützt ggf.)
@@ -137,21 +131,46 @@ func Register(mux *http.ServeMux, d Deps) {
http.NotFound(w, r)
return
}
id, err := strconv.ParseInt(parts[0], 10, 64)
id := parts[0]
if strings.TrimSpace(id) == "" {
http.NotFound(w, r)
return
}
// 1) lokal versuchen
if rc, meta, err := d.Blob.Open(r.Context(), id); err == nil {
defer rc.Close()
w.Header().Set("Content-Type", meta.ContentType)
w.Header().Set("Content-Length", strconv.FormatInt(meta.Size, 10))
w.Header().Set("Content-Disposition", fmt.Sprintf(`attachment; filename="%s"`, meta.Name))
_, _ = io.Copy(w, rc)
return
}
// 2) Remote über Mesh holen
rrc, name, _, _, err := d.Mesh.FetchBlobAny(r.Context(), id)
if err != nil {
http.NotFound(w, r)
return
}
rc, meta, err := d.Blob.Open(r.Context(), id)
if err != nil {
http.NotFound(w, r)
defer rrc.Close()
// 3) lokal cachen (Save konsumiert den Stream)
if _, err := d.Blob.Save(r.Context(), id, name, rrc); err != nil {
http.Error(w, "cache failed: "+err.Error(), http.StatusInternalServerError)
return
}
defer rc.Close()
// 4) aus lokalem Store ausliefern (saubere Größe/CT)
lrc, meta, err := d.Blob.Open(r.Context(), id)
if err != nil {
http.Error(w, "open failed", http.StatusInternalServerError)
return
}
defer lrc.Close()
w.Header().Set("Content-Type", meta.ContentType)
w.Header().Set("Content-Length", strconv.FormatInt(meta.Size, 10))
w.Header().Set("Content-Disposition", fmt.Sprintf(`attachment; filename="%s"`, meta.Name))
_, _ = io.Copy(w, rc)
_, _ = io.Copy(w, lrc)
})
mux.HandleFunc("/admin/peers", func(w http.ResponseWriter, r *http.Request) {
@@ -162,7 +181,7 @@ func Register(mux *http.ServeMux, d Deps) {
})
})
// Actions (HTMX POSTs)
// CREATE
mux.HandleFunc("/admin/items/create", func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
@@ -171,36 +190,40 @@ func Register(mux *http.ServeMux, d Deps) {
name := strings.TrimSpace(r.FormValue("name"))
if name != "" {
_, _ = d.Store.Create(r.Context(), name)
_ = d.Mesh.SyncNow(r.Context()) // prompt push (best effort)
_ = d.Mesh.SyncNow(r.Context())
}
// Nach Aktion Items partial zurückgeben (HTMX swap)
http.Redirect(w, r, "/admin/items", http.StatusSeeOther)
// statt Redirect:
renderItemsPartial(w, r, d)
})
// RENAME
mux.HandleFunc("/admin/items/rename", func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
idStr := r.FormValue("id")
id := strings.TrimSpace(r.FormValue("id"))
newName := strings.TrimSpace(r.FormValue("name"))
if id, err := strconv.ParseInt(idStr, 10, 64); err == nil && newName != "" {
if id != "" && newName != "" {
_, _ = d.Store.Rename(r.Context(), filesvc.ID(id), newName)
_ = d.Mesh.SyncNow(r.Context())
}
http.Redirect(w, r, "/admin/items", http.StatusSeeOther)
renderItemsPartial(w, r, d)
})
// DELETE
mux.HandleFunc("/admin/items/delete", func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
if id, err := strconv.ParseInt(r.FormValue("id"), 10, 64); err == nil {
id := strings.TrimSpace(r.FormValue("id"))
if id != "" {
_, _ = d.Store.Delete(r.Context(), filesvc.ID(id))
_ = d.Blob.Delete(r.Context(), id)
_ = d.Mesh.SyncNow(r.Context())
}
http.Redirect(w, r, "/admin/items", http.StatusSeeOther)
renderItemsPartial(w, r, d)
})
mux.HandleFunc("/admin/mesh/syncnow", func(w http.ResponseWriter, r *http.Request) {
@@ -231,3 +254,34 @@ func BasicAuth(user, pass string, next http.Handler) http.Handler {
next.ServeHTTP(w, r)
})
}
// rebuild & render items partial for HTMX swaps
func renderItemsPartial(w http.ResponseWriter, r *http.Request, d Deps) {
type row struct {
ID string
Name string
UpdatedAt int64
HasBlob bool
Size int64
}
nextID := filesvc.ID(strings.TrimSpace(r.URL.Query().Get("next")))
items, nextOut, _ := d.Store.List(r.Context(), nextID, 100)
rows := make([]row, 0, len(items))
for _, it := range items {
meta, ok, _ := d.Blob.Stat(r.Context(), (it.ID))
rows = append(rows, row{
ID: (it.ID),
Name: it.Name,
UpdatedAt: it.UpdatedAt,
HasBlob: ok,
Size: meta.Size,
})
}
w.Header().Set("Content-Type", "text/html; charset=utf-8")
_ = tplItems.Execute(w, map[string]any{
"Items": rows,
"Next": nextOut,
})
}

View File

@@ -8,7 +8,7 @@
<style>
:root { --bg:#0b1220; --card:#121a2b; --muted:#94a3b8; --text:#e5e7eb; --accent:#4f46e5; }
html,body { margin:0; background:var(--bg); color:var(--text); font-family: ui-sans-serif, system-ui, -apple-system, Segoe UI, Roboto, Ubuntu; }
.wrap { max-width: 980px; margin: 40px auto; padding: 0 16px; }
.wrap { max-width: 1200px; margin: 40px auto; padding: 0 16px; }
.nav { display:flex; gap:12px; margin-bottom:16px; }
.btn { background:var(--card); border:1px solid #243044; padding:8px 12px; border-radius:10px; color:var(--text); cursor:pointer; }
.btn:hover { border-color:#3b4a66; }

View File

@@ -1,6 +1,6 @@
<div class="row">
<div style="flex: 1 1 360px">
<form hx-post="/admin/items/create" hx-target="#items" hx-get="/admin/items" hx-swap="outerHTML">
<form hx-post="/admin/items/create" hx-target="#items" hx-swap="outerHTML">
<label>Neue leere Datei (nur Metadaten)</label>
<div style="display:flex; gap:8px; margin-top:6px">
<input type="text" name="name" placeholder="z.B. notes.txt" required>
@@ -42,17 +42,17 @@
{{ end }}
</td>
<td>
<form style="display:inline-flex; gap:6px"
hx-post="/admin/items/rename"
hx-target="#items" hx-get="/admin/items" hx-swap="outerHTML">
<form style="display:inline-flex; gap:6px"
hx-post="/admin/items/rename"
hx-target="#items" hx-swap="outerHTML">
<input type="hidden" name="id" value="{{ .ID }}">
<input type="text" name="name" placeholder="Neuer Name">
<button class="btn" type="submit">Rename</button>
</form>
<form style="display:inline"
hx-post="/admin/items/delete"
hx-target="#items" hx-get="/admin/items" hx-swap="outerHTML"
onsubmit="return confirm('Wirklich löschen (inkl. Blob)?');">
<form style="display:inline"
hx-post="/admin/items/delete"
hx-target="#items" hx-swap="outerHTML"
onsubmit="return confirm('Wirklich löschen (inkl. Blob)?');">
<input type="hidden" name="id" value="{{ .ID }}">
<button class="btn" type="submit">Delete</button>
</form>

View File

@@ -11,11 +11,11 @@
{{ range .Peers }}
<tr>
<td>{{ .URL }}</td>
<td>{{ if .Self }}✅{{ else }}{{ end }}</td>
<td>{{ if .Self }}✅{{ else }}🔗{{ end }}</td>
<td><small class="muted">{{ .LastSeen }}</small></td>
</tr>
{{ else }}
<tr><td colspan="3" class="muted">Keine Peers vorhanden.</td></tr>
<tr><td colspan="3" class="muted">⚠️ Keine Peers vorhanden.</td></tr>
{{ end }}
</tbody>
</table>

View File

@@ -6,7 +6,6 @@ import (
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"io"
"mime"
"net/http"
@@ -23,22 +22,35 @@ type Meta struct {
}
type Store interface {
Save(ctx context.Context, id int64, filename string, r io.Reader) (Meta, error)
Open(ctx context.Context, id int64) (io.ReadSeekCloser, Meta, error)
Stat(ctx context.Context, id int64) (Meta, bool, error)
Delete(ctx context.Context, id int64) error
Save(ctx context.Context, id string, filename string, r io.Reader) (Meta, error)
Open(ctx context.Context, id string) (io.ReadSeekCloser, Meta, error)
Stat(ctx context.Context, id string) (Meta, bool, error)
Delete(ctx context.Context, id string) error
}
type FS struct{ root string }
func New(root string) *FS { return &FS{root: root} }
func (fs *FS) dir(id int64) string { return filepath.Join(fs.root, "files", fmt.Sprintf("%d", id)) }
func (fs *FS) metaPath(id int64) string { return filepath.Join(fs.dir(id), "meta.json") }
func (fs *FS) blobPath(id int64, name string) string {
func (fs *FS) dir(id string) string { return filepath.Join(fs.root, "files", sanitizeID(id)) }
func (fs *FS) metaPath(id string) string { return filepath.Join(fs.dir(id), "meta.json") }
func (fs *FS) blobPath(id string, name string) string {
return filepath.Join(fs.dir(id), "blob"+safeExt(name))
}
func sanitizeID(id string) string {
// nur 0-9a-zA-Z- zulassen; Rest mit '_' ersetzen
b := make([]rune, 0, len(id))
for _, r := range id {
if (r >= '0' && r <= '9') || (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || r == '-' {
b = append(b, r)
} else {
b = append(b, '_')
}
}
return string(b)
}
func safeExt(name string) string {
ext := filepath.Ext(name)
if len(ext) > 16 { // unrealistisch lange Exts beschneiden
@@ -47,7 +59,7 @@ func safeExt(name string) string {
return ext
}
func (fs *FS) Save(_ context.Context, id int64, filename string, r io.Reader) (Meta, error) {
func (fs *FS) Save(_ context.Context, id string, filename string, r io.Reader) (Meta, error) {
if strings.TrimSpace(filename) == "" {
return Meta{}, errors.New("filename required")
}
@@ -107,7 +119,7 @@ func (fs *FS) Save(_ context.Context, id int64, filename string, r io.Reader) (M
return meta, nil
}
func (fs *FS) Open(_ context.Context, id int64) (io.ReadSeekCloser, Meta, error) {
func (fs *FS) Open(_ context.Context, id string) (io.ReadSeekCloser, Meta, error) {
meta, ok, err := fs.Stat(context.Background(), id)
if err != nil {
return nil, Meta{}, err
@@ -119,7 +131,7 @@ func (fs *FS) Open(_ context.Context, id int64) (io.ReadSeekCloser, Meta, error)
return f, meta, err
}
func (fs *FS) Stat(_ context.Context, id int64) (Meta, bool, error) {
func (fs *FS) Stat(_ context.Context, id string) (Meta, bool, error) {
b, err := os.ReadFile(fs.metaPath(id))
if err != nil {
if os.IsNotExist(err) {
@@ -139,7 +151,7 @@ func (fs *FS) Stat(_ context.Context, id int64) (Meta, bool, error) {
return m, true, nil
}
func (fs *FS) Delete(_ context.Context, id int64) error {
func (fs *FS) Delete(_ context.Context, id string) error {
return os.RemoveAll(fs.dir(id))
}

View File

@@ -11,7 +11,6 @@ import (
type MemStore struct {
mu sync.Mutex
items map[ID]File
next ID
// optionales Eventing
subs []chan ChangeEvent
@@ -20,7 +19,6 @@ type MemStore struct {
func NewMemStore() *MemStore {
return &MemStore{
items: make(map[ID]File),
next: 1,
}
}
@@ -39,39 +37,54 @@ func (m *MemStore) Get(_ context.Context, id ID) (File, error) {
func (m *MemStore) List(_ context.Context, next ID, limit int) ([]File, ID, error) {
m.mu.Lock()
defer m.mu.Unlock()
if limit <= 0 || limit > 1000 {
limit = 100
}
ids := make([]ID, 0, len(m.items))
for id := range m.items {
ids = append(ids, id)
// sortiere deterministisch nach UpdatedAt, dann ID
all := make([]File, 0, len(m.items))
for _, v := range m.items {
all = append(all, v)
}
slices.Sort(ids)
slices.SortFunc(all, func(a, b File) int {
if a.UpdatedAt == b.UpdatedAt {
if a.ID == b.ID {
return 0
}
if a.ID < b.ID {
return -1
}
return 1
}
if a.UpdatedAt < b.UpdatedAt {
return -1
}
return 1
})
start := 0
if next > 0 {
for i, id := range ids {
if id >= next {
if next != "" {
for i, it := range all {
if it.ID >= next {
start = i
break
}
}
}
end := start + limit
if end > len(ids) {
end = len(ids)
if end > len(all) {
end = len(all)
}
out := make([]File, 0, end-start)
for _, id := range ids[start:end] {
it := m.items[id]
for _, it := range all[start:end] {
if !it.Deleted {
out = append(out, it)
}
}
var nextOut ID
if end < len(ids) {
nextOut = ids[end]
if end < len(all) {
nextOut = all[end].ID
}
return out, nextOut, nil
}
@@ -85,9 +98,12 @@ func (m *MemStore) Create(_ context.Context, name string) (File, error) {
m.mu.Lock()
defer m.mu.Unlock()
now := time.Now().UnixNano()
it := File{ID: m.next, Name: name, UpdatedAt: now}
uid, err := NewUUIDv4()
if err != nil {
return File{}, err
}
it := File{ID: uid, Name: name, UpdatedAt: now}
m.items[it.ID] = it
m.next++
m.emit(it)
return it, nil
}
@@ -147,9 +163,6 @@ func (m *MemStore) ApplyRemote(_ context.Context, s Snapshot) error {
li, ok := m.items[ri.ID]
if !ok || ri.UpdatedAt > li.UpdatedAt {
m.items[ri.ID] = ri
if ri.ID >= m.next {
m.next = ri.ID + 1
}
m.emitLocked(ri)
}
}

View File

@@ -2,13 +2,15 @@ package filesvc
import (
"context"
"crypto/rand"
"errors"
"fmt"
"time"
)
/*** Domain ***/
type ID = int64
type ID = string
type File struct {
ID ID `json:"id"`
@@ -76,3 +78,13 @@ type MeshStore interface {
Replicable
Watchable // optional kann Noop sein
}
func NewUUIDv4() (string, error) {
b := make([]byte, 16)
if _, err := rand.Read(b); err != nil {
return "", err
}
b[6] = (b[6] & 0x0f) | 0x40 // Version 4
b[8] = (b[8] & 0x3f) | 0x80 // Variant RFC4122
return fmt.Sprintf("%x-%x-%x-%x-%x", b[0:4], b[4:6], b[6:8], b[8:10], b[10:16]), nil
}

View File

@@ -8,6 +8,7 @@ import (
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"hash/crc32"
"io"
"log"
@@ -15,6 +16,7 @@ import (
"net/http"
"os"
"slices"
"strconv"
"strings"
"sync"
"time"
@@ -28,7 +30,9 @@ type Config struct {
Seeds []string // other peers' mesh base URLs
ClusterSecret string // HMAC key
EnableDiscovery bool
DiscoveryAddress string // "239.8.8.8:9898"
DiscoveryAddress string // "239.8.8.8:9898"
PeerTTL time.Duration // wie lange darf ein Peer inaktiv sein (Default siehe unten)
PruneInterval time.Duration // wie oft wird gepruned
}
type Peer struct {
@@ -39,7 +43,7 @@ type Peer struct {
}
type Item struct {
ID int64 `json:"id"`
ID string `json:"id"`
Name string `json:"name"`
UpdatedAt int64 `json:"updatedAt"`
Deleted bool `json:"deleted"` // <— NEU: Tombstone für Deletes
@@ -53,6 +57,7 @@ type Snapshot struct {
type Callbacks struct {
GetSnapshot func(ctx context.Context) (Snapshot, error)
ApplyRemote func(ctx context.Context, s Snapshot) error
BlobOpen func(ctx context.Context, id string) (io.ReadCloser, string, string, int64, error)
}
/*** Node ***/
@@ -69,6 +74,78 @@ type Node struct {
wg sync.WaitGroup
}
// RemovePeer löscht einen Peer aus der Peer-Tabelle. Seeds werden standardmäßig nicht entfernt.
func (n *Node) RemovePeer(url string) bool {
n.mu.Lock()
defer n.mu.Unlock()
if url == "" || url == n.self.URL {
return false
}
// Seeds schützen
if n.isSeed(url) {
return false
}
if _, ok := n.peers[url]; ok {
delete(n.peers, url)
return true
}
return false
}
// PruneNow entfernt alle Peers, deren LastSeen vor cutoff liegt (Seeds bleiben).
func (n *Node) PruneNow(cutoff time.Time) int {
n.mu.Lock()
defer n.mu.Unlock()
removed := 0
for url, p := range n.peers {
if url == n.self.URL || n.isSeed(url) {
continue
}
if p.LastSeen.IsZero() || p.LastSeen.Before(cutoff) {
delete(n.peers, url)
removed++
}
}
return removed
}
// StartPeerPruner startet den Hintergrundjob (stoppt automatisch bei n.stop).
func (n *Node) StartPeerPruner() {
go n.loopPrunePeers()
}
func (n *Node) loopPrunePeers() {
ttl := n.cfg.PeerTTL
if ttl <= 0 {
ttl = 2 * time.Minute
}
interval := n.cfg.PruneInterval
if interval <= 0 {
interval = 30 * time.Second
}
t := time.NewTicker(interval)
defer t.Stop()
for {
select {
case <-n.stop:
return
case <-t.C:
cutoff := time.Now().Add(-ttl)
_ = n.PruneNow(cutoff)
}
}
}
// helper: ist url ein Seed?
func (n *Node) isSeed(url string) bool {
for _, s := range n.cfg.Seeds {
if strings.TrimSpace(s) == strings.TrimSpace(url) {
return true
}
}
return false
}
func New(cfg Config, cbs Callbacks) (*Node, error) {
if cfg.BindAddr == "" || cfg.AdvertURL == "" {
return nil, errors.New("mesh: BindAddr and AdvertURL required")
@@ -133,6 +210,7 @@ func (n *Node) peersHandler(w http.ResponseWriter, r *http.Request) {
var list []Peer
list = append(list, n.self)
for _, p := range n.peers {
//p.Self = false
list = append(list, *p)
}
writeJSON(w, http.StatusOK, list)
@@ -170,6 +248,7 @@ func (n *Node) Serve() error {
mux := http.NewServeMux()
mux.HandleFunc("/mesh/peers", n.peersHandler)
mux.HandleFunc("/mesh/hello", n.helloHandler)
mux.HandleFunc("/mesh/blob", n.blobHandler)
mux.HandleFunc("/mesh/sync", n.syncHandler)
n.srv = &http.Server{Addr: n.cfg.BindAddr, Handler: mux}
@@ -180,6 +259,9 @@ func (n *Node) Serve() error {
n.loopSeeder()
}()
n.wg.Add(1)
go func() { defer n.wg.Done(); n.loopPeerExchange() }()
if n.cfg.EnableDiscovery && n.cfg.DiscoveryAddress != "" {
n.wg.Add(2)
go func() {
@@ -222,6 +304,43 @@ func (n *Node) Close(ctx context.Context) error {
/*** Loops ***/
func (n *Node) loopPeerExchange() {
t := time.NewTicker(30 * time.Second)
defer t.Stop()
for {
select {
case <-n.stop:
return
case <-t.C:
}
// Seeds abfragen
for _, s := range n.cfg.Seeds {
if strings.TrimSpace(s) == "" {
continue
}
resp, err := n.client.Get(strings.TrimRight(s, "/") + "/mesh/peers")
if err != nil {
continue
}
var list []Peer
if json.NewDecoder(resp.Body).Decode(&list) == nil {
n.mu.Lock()
for _, p := range list {
if p.URL == "" || p.URL == n.self.URL {
continue
}
if _, ok := n.peers[p.URL]; !ok {
cp := p
n.peers[p.URL] = &cp
}
}
n.mu.Unlock()
}
resp.Body.Close()
}
}
}
func (n *Node) loopSeeder() {
// attempt to hello known seeds every 5s at start, then every 30s
backoff := 5 * time.Second
@@ -377,6 +496,7 @@ func (n *Node) PeerList() []Peer {
out = append(out, n.self)
for _, p := range n.peers {
cp := *p
cp.Self = false
out = append(out, cp)
}
return out
@@ -403,11 +523,11 @@ func (n *Node) SyncNow(ctx context.Context) error {
/*** Utilities ***/
// OwnerHint is a simple, optional mapping to distribute responsibility.
func OwnerHint(id int64, peers []string) int {
func OwnerHint(id string, peers []string) int {
if len(peers) == 0 {
return 0
}
h := crc32.ChecksumIEEE([]byte(string(rune(id))))
h := crc32.ChecksumIEEE([]byte(id))
return int(h % uint32(len(peers)))
}
@@ -449,3 +569,86 @@ func getenvDefault(k, def string) string {
}
return v
}
// POST /mesh/blob (Body: {"id":<int64>}) -> streamt den Blob
func (n *Node) blobHandler(w http.ResponseWriter, r *http.Request) {
body, _ := io.ReadAll(r.Body)
if !n.verify(body, r.Header.Get("X-Mesh-Sig")) {
http.Error(w, "bad signature", http.StatusUnauthorized)
return
}
var req struct {
ID string `json:"id"`
}
if err := json.Unmarshal(body, &req); err != nil {
http.Error(w, "bad json", http.StatusBadRequest)
return
}
if n.cbs.BlobOpen == nil {
http.Error(w, "blob unavailable", http.StatusNotFound)
return
}
rc, name, ct, size, err := n.cbs.BlobOpen(r.Context(), req.ID)
if err != nil {
http.Error(w, "not found", http.StatusNotFound)
return
}
defer rc.Close()
if ct == "" {
ct = "application/octet-stream"
}
if size > 0 {
w.Header().Set("Content-Length", strconv.FormatInt(size, 10))
}
w.Header().Set("Content-Type", ct)
w.Header().Set("X-Blob-Name", name)
_, _ = io.Copy(w, rc)
}
// interner Helper: signierter Blob-Request an einen Peer
func (n *Node) sendBlobRequest(url string, id string) (*http.Response, error) {
b, _ := json.Marshal(struct {
ID string `json:"id"`
}{ID: id})
req, _ := http.NewRequest(http.MethodPost, strings.TrimRight(url, "/")+"/mesh/blob", bytes.NewReader(b))
req.Header.Set("X-Mesh-Sig", n.sign(b))
return n.client.Do(req)
}
// Öffentliche Methode: versuche Blob bei irgendeinem Peer zu holen
func (n *Node) FetchBlobAny(ctx context.Context, id string) (io.ReadCloser, string, string, int64, error) {
n.mu.RLock()
targets := make([]string, 0, len(n.peers))
for url := range n.peers {
targets = append(targets, url)
}
n.mu.RUnlock()
if len(targets) == 0 {
// Fallback: Seeds probieren
targets = append(targets, n.cfg.Seeds...)
}
for _, u := range targets {
if strings.TrimSpace(u) == "" || u == n.self.URL {
continue
}
resp, err := n.sendBlobRequest(u, id)
if err != nil {
continue
}
if resp.StatusCode == http.StatusOK {
name := resp.Header.Get("X-Blob-Name")
ct := resp.Header.Get("Content-Type")
var size int64 = -1
if cl := resp.Header.Get("Content-Length"); cl != "" {
if s, err := strconv.ParseInt(cl, 10, 64); err == nil {
size = s
}
}
// Caller muss resp.Body schließen
return resp.Body, name, ct, size, nil
}
io.Copy(io.Discard, resp.Body)
resp.Body.Close()
}
return nil, "", "", 0, fmt.Errorf("blob %s not found on peers", id)
}