feat: scaffold files service
This commit is contained in:
62
internal/config/config.go
Normal file
62
internal/config/config.go
Normal file
@@ -0,0 +1,62 @@
|
||||
package config
|
||||
|
||||
import (
|
||||
"os"
|
||||
"strconv"
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
ServerPort string
|
||||
DatabaseURL string
|
||||
MigrationsDir string
|
||||
InternalAPIKey string
|
||||
PublicBaseURL string
|
||||
|
||||
MinIOEndpoint string
|
||||
MinIOAccessKey string
|
||||
MinIOSecretKey string
|
||||
MinIOBucket string
|
||||
MinIOUseSSL bool
|
||||
|
||||
PodName string
|
||||
}
|
||||
|
||||
func Load() *Config {
|
||||
return &Config{
|
||||
ServerPort: envStr("SERVER_PORT", "3001"),
|
||||
DatabaseURL: envStr("DATABASE_URL", "postgres://files:files@localhost:5432/files?sslmode=disable"),
|
||||
MigrationsDir: envStr("MIGRATIONS_DIR", "/migrations"),
|
||||
InternalAPIKey: envStr("INTERNAL_API_KEY", envStr("PORTAL_INTERNAL_API_KEY", "")),
|
||||
PublicBaseURL: envStr("PUBLIC_BASE_URL", "https://portal.estateliga.work"),
|
||||
MinIOEndpoint: envStr("MINIO_ENDPOINT", ""),
|
||||
MinIOAccessKey: envStr("MINIO_ACCESS_KEY", ""),
|
||||
MinIOSecretKey: envStr("MINIO_SECRET_KEY", ""),
|
||||
MinIOBucket: envStr("MINIO_BUCKET", "portal-files"),
|
||||
MinIOUseSSL: envBool("MINIO_USE_SSL", false),
|
||||
PodName: envStr("POD_NAME", hostname()),
|
||||
}
|
||||
}
|
||||
|
||||
func envBool(key string, def bool) bool {
|
||||
if v := os.Getenv(key); v != "" {
|
||||
if b, err := strconv.ParseBool(v); err == nil {
|
||||
return b
|
||||
}
|
||||
}
|
||||
return def
|
||||
}
|
||||
|
||||
func envStr(key, def string) string {
|
||||
if v := os.Getenv(key); v != "" {
|
||||
return v
|
||||
}
|
||||
return def
|
||||
}
|
||||
|
||||
func hostname() string {
|
||||
h, err := os.Hostname()
|
||||
if err != nil {
|
||||
return "unknown"
|
||||
}
|
||||
return h
|
||||
}
|
||||
27
internal/handler/health.go
Normal file
27
internal/handler/health.go
Normal file
@@ -0,0 +1,27 @@
|
||||
package handler
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
type HealthHandler struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewHealthHandler(pool *pgxpool.Pool) *HealthHandler {
|
||||
return &HealthHandler{pool: pool}
|
||||
}
|
||||
|
||||
func (h *HealthHandler) Healthz(w http.ResponseWriter, _ *http.Request) {
|
||||
writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
|
||||
}
|
||||
|
||||
func (h *HealthHandler) Readyz(w http.ResponseWriter, r *http.Request) {
|
||||
if err := h.pool.Ping(r.Context()); err != nil {
|
||||
writeInternalError(w, r, err, "database unavailable")
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, map[string]string{"status": "ready"})
|
||||
}
|
||||
44
internal/handler/helpers.go
Normal file
44
internal/handler/helpers.go
Normal file
@@ -0,0 +1,44 @@
|
||||
package handler
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"strings"
|
||||
)
|
||||
|
||||
func writeJSON(w http.ResponseWriter, status int, v any) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(status)
|
||||
_ = json.NewEncoder(w).Encode(v)
|
||||
}
|
||||
|
||||
func writeError(w http.ResponseWriter, status int, msg string) {
|
||||
writeJSON(w, status, map[string]string{"error": msg})
|
||||
}
|
||||
|
||||
func writeInternalError(w http.ResponseWriter, r *http.Request, err error, msg string) {
|
||||
slog.Error("http error", "method", r.Method, "path", r.URL.Path, "err", err)
|
||||
writeError(w, http.StatusInternalServerError, msg)
|
||||
}
|
||||
|
||||
func decodeJSON(r *http.Request, v any) error {
|
||||
defer r.Body.Close()
|
||||
return json.NewDecoder(r.Body).Decode(v)
|
||||
}
|
||||
|
||||
func csvHeader(r *http.Request, key string) []string {
|
||||
raw := r.Header.Get(key)
|
||||
if raw == "" {
|
||||
return nil
|
||||
}
|
||||
parts := strings.Split(raw, ",")
|
||||
out := make([]string, 0, len(parts))
|
||||
for _, p := range parts {
|
||||
p = strings.TrimSpace(p)
|
||||
if p != "" {
|
||||
out = append(out, p)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
335
internal/handler/node.go
Normal file
335
internal/handler/node.go
Normal file
@@ -0,0 +1,335 @@
|
||||
package handler
|
||||
|
||||
import (
|
||||
"crypto/rand"
|
||||
"encoding/base64"
|
||||
"errors"
|
||||
"io"
|
||||
"net/http"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
commonmw "gitea.estateliga.work/admin/portal-common/middleware"
|
||||
"github.com/go-chi/chi/v5"
|
||||
|
||||
"files-service/internal/config"
|
||||
"files-service/internal/model"
|
||||
"files-service/internal/repository"
|
||||
"files-service/internal/storage"
|
||||
)
|
||||
|
||||
type NodeHandler struct {
|
||||
cfg *config.Config
|
||||
repo *repository.NodeRepository
|
||||
store *storage.Storage
|
||||
}
|
||||
|
||||
func NewNodeHandler(cfg *config.Config, repo *repository.NodeRepository, store *storage.Storage) *NodeHandler {
|
||||
return &NodeHandler{cfg: cfg, repo: repo, store: store}
|
||||
}
|
||||
|
||||
func (h *NodeHandler) List(w http.ResponseWriter, r *http.Request) {
|
||||
userID := commonmw.GetUserID(r.Context())
|
||||
scope := r.URL.Query().Get("scope")
|
||||
parentID := emptyToNil(r.URL.Query().Get("parent_id"))
|
||||
nodes, err := h.repo.List(r.Context(), userID, subordinates(r), scope, parentID)
|
||||
if err != nil {
|
||||
writeInternalError(w, r, err, "failed to list files")
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, nodes)
|
||||
}
|
||||
|
||||
func (h *NodeHandler) Get(w http.ResponseWriter, r *http.Request) {
|
||||
node, ok := h.requireNode(w, r)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, node)
|
||||
}
|
||||
|
||||
func (h *NodeHandler) CreateFolder(w http.ResponseWriter, r *http.Request) {
|
||||
var req model.CreateFolderRequest
|
||||
if err := decodeJSON(r, &req); err != nil {
|
||||
writeError(w, http.StatusBadRequest, "invalid json")
|
||||
return
|
||||
}
|
||||
req.Title = strings.TrimSpace(req.Title)
|
||||
if req.Title == "" {
|
||||
writeError(w, http.StatusBadRequest, "title is required")
|
||||
return
|
||||
}
|
||||
userID := commonmw.GetUserID(r.Context())
|
||||
node, err := h.repo.CreateFolder(r.Context(), req.Title, req.ParentID, userID)
|
||||
if err != nil {
|
||||
writeInternalError(w, r, err, "failed to create folder")
|
||||
return
|
||||
}
|
||||
h.repo.Audit(r.Context(), userID, "files.folder_create", "files_node", node.ID, "{}")
|
||||
writeJSON(w, http.StatusCreated, node)
|
||||
}
|
||||
|
||||
func (h *NodeHandler) UploadFile(w http.ResponseWriter, r *http.Request) {
|
||||
if !h.store.Configured() {
|
||||
writeError(w, http.StatusServiceUnavailable, "storage not configured")
|
||||
return
|
||||
}
|
||||
if err := r.ParseMultipartForm(64 << 20); err != nil {
|
||||
writeError(w, http.StatusBadRequest, "invalid multipart")
|
||||
return
|
||||
}
|
||||
file, header, err := r.FormFile("file")
|
||||
if err != nil {
|
||||
writeError(w, http.StatusBadRequest, "file is required")
|
||||
return
|
||||
}
|
||||
defer file.Close()
|
||||
|
||||
filename := filepath.Base(header.Filename)
|
||||
ext := strings.ToLower(strings.TrimPrefix(filepath.Ext(filename), "."))
|
||||
if !storage.AllowedExtension(ext) {
|
||||
writeError(w, http.StatusBadRequest, "file type is not allowed")
|
||||
return
|
||||
}
|
||||
title := strings.TrimSpace(r.FormValue("title"))
|
||||
if title == "" {
|
||||
title = strings.TrimSuffix(filename, filepath.Ext(filename))
|
||||
}
|
||||
userID := commonmw.GetUserID(r.Context())
|
||||
key := storage.GenerateKey(userID, filename)
|
||||
contentType := storage.GuessContentType(filename, header.Header.Get("Content-Type"))
|
||||
if err := h.store.PutObject(r.Context(), key, file, header.Size, contentType); err != nil {
|
||||
writeInternalError(w, r, err, "failed to upload file")
|
||||
return
|
||||
}
|
||||
parentID := emptyToNil(r.FormValue("parent_id"))
|
||||
node, err := h.repo.CreateFile(r.Context(), &model.Node{
|
||||
ParentID: parentID,
|
||||
Title: title,
|
||||
OwnerUserID: userID,
|
||||
StorageKey: &key,
|
||||
OriginalFilename: &filename,
|
||||
MimeType: &contentType,
|
||||
Extension: &ext,
|
||||
SizeBytes: header.Size,
|
||||
})
|
||||
if err != nil {
|
||||
writeInternalError(w, r, err, "failed to create file")
|
||||
return
|
||||
}
|
||||
h.repo.Audit(r.Context(), userID, "files.file_upload", "files_node", node.ID, "{}")
|
||||
writeJSON(w, http.StatusCreated, node)
|
||||
}
|
||||
|
||||
func (h *NodeHandler) Update(w http.ResponseWriter, r *http.Request) {
|
||||
var req model.UpdateNodeRequest
|
||||
if err := decodeJSON(r, &req); err != nil {
|
||||
writeError(w, http.StatusBadRequest, "invalid json")
|
||||
return
|
||||
}
|
||||
if req.Title != nil {
|
||||
title := strings.TrimSpace(*req.Title)
|
||||
if title == "" {
|
||||
writeError(w, http.StatusBadRequest, "title is required")
|
||||
return
|
||||
}
|
||||
req.Title = &title
|
||||
}
|
||||
userID := commonmw.GetUserID(r.Context())
|
||||
node, err := h.repo.Update(r.Context(), chi.URLParam(r, "id"), userID, req)
|
||||
if errors.Is(err, repository.ErrNotFound) {
|
||||
writeError(w, http.StatusNotFound, "file not found")
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
writeInternalError(w, r, err, "failed to update file")
|
||||
return
|
||||
}
|
||||
h.repo.Audit(r.Context(), userID, "files.node_update", "files_node", node.ID, "{}")
|
||||
writeJSON(w, http.StatusOK, node)
|
||||
}
|
||||
|
||||
func (h *NodeHandler) Delete(w http.ResponseWriter, r *http.Request) {
|
||||
userID := commonmw.GetUserID(r.Context())
|
||||
id := chi.URLParam(r, "id")
|
||||
if err := h.repo.SoftDelete(r.Context(), id, userID); errors.Is(err, repository.ErrNotFound) {
|
||||
writeError(w, http.StatusNotFound, "file not found")
|
||||
return
|
||||
} else if err != nil {
|
||||
writeInternalError(w, r, err, "failed to delete file")
|
||||
return
|
||||
}
|
||||
h.repo.Audit(r.Context(), userID, "files.node_delete", "files_node", id, "{}")
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
}
|
||||
|
||||
func (h *NodeHandler) Download(w http.ResponseWriter, r *http.Request) {
|
||||
node, ok := h.requireNode(w, r)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
h.streamNode(w, r, node)
|
||||
}
|
||||
|
||||
func (h *NodeHandler) ListAccess(w http.ResponseWriter, r *http.Request) {
|
||||
node, ok := h.requireNode(w, r)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if node.EffectiveAccess != model.AccessEdit {
|
||||
writeError(w, http.StatusForbidden, "edit access required")
|
||||
return
|
||||
}
|
||||
access, err := h.repo.ListAccess(r.Context(), node.ID)
|
||||
if err != nil {
|
||||
writeInternalError(w, r, err, "failed to list access")
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, access)
|
||||
}
|
||||
|
||||
func (h *NodeHandler) ReplaceAccess(w http.ResponseWriter, r *http.Request) {
|
||||
var req model.ReplaceAccessRequest
|
||||
if err := decodeJSON(r, &req); err != nil {
|
||||
writeError(w, http.StatusBadRequest, "invalid json")
|
||||
return
|
||||
}
|
||||
userID := commonmw.GetUserID(r.Context())
|
||||
id := chi.URLParam(r, "id")
|
||||
if err := h.repo.ReplaceAccess(r.Context(), id, userID, req.Access); errors.Is(err, repository.ErrNotFound) {
|
||||
writeError(w, http.StatusNotFound, "file not found")
|
||||
return
|
||||
} else if err != nil {
|
||||
writeInternalError(w, r, err, "failed to update access")
|
||||
return
|
||||
}
|
||||
h.repo.Audit(r.Context(), userID, "files.access_update", "files_node", id, "{}")
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
}
|
||||
|
||||
func (h *NodeHandler) CreatePublicLink(w http.ResponseWriter, r *http.Request) {
|
||||
var req model.PublicLinkRequest
|
||||
if err := decodeJSON(r, &req); err != nil {
|
||||
writeError(w, http.StatusBadRequest, "invalid json")
|
||||
return
|
||||
}
|
||||
if req.ExpiresAt.Before(time.Now().Add(time.Minute)) {
|
||||
writeError(w, http.StatusBadRequest, "expires_at must be in future")
|
||||
return
|
||||
}
|
||||
token, err := newToken()
|
||||
if err != nil {
|
||||
writeInternalError(w, r, err, "failed to create public link")
|
||||
return
|
||||
}
|
||||
userID := commonmw.GetUserID(r.Context())
|
||||
id := chi.URLParam(r, "id")
|
||||
linkID, err := h.repo.CreatePublicLink(r.Context(), id, userID, token, req.ExpiresAt)
|
||||
if errors.Is(err, repository.ErrNotFound) {
|
||||
writeError(w, http.StatusNotFound, "file not found")
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
writeInternalError(w, r, err, "failed to create public link")
|
||||
return
|
||||
}
|
||||
h.repo.Audit(r.Context(), userID, "files.public_link_create", "files_node", id, "{}")
|
||||
writeJSON(w, http.StatusCreated, model.PublicLinkResponse{
|
||||
ID: linkID,
|
||||
URL: strings.TrimRight(h.cfg.PublicBaseURL, "/") + "/public/files/" + token,
|
||||
ExpiresAt: req.ExpiresAt,
|
||||
})
|
||||
}
|
||||
|
||||
func (h *NodeHandler) PublicMeta(w http.ResponseWriter, r *http.Request) {
|
||||
node, ok := h.publicNode(w, r)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, node)
|
||||
}
|
||||
|
||||
func (h *NodeHandler) PublicDownload(w http.ResponseWriter, r *http.Request) {
|
||||
node, ok := h.publicNode(w, r)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
h.streamNode(w, r, node)
|
||||
}
|
||||
|
||||
func (h *NodeHandler) requireNode(w http.ResponseWriter, r *http.Request) (*model.Node, bool) {
|
||||
node, err := h.repo.GetForUser(r.Context(), chi.URLParam(r, "id"), commonmw.GetUserID(r.Context()), subordinates(r))
|
||||
if errors.Is(err, repository.ErrNotFound) {
|
||||
writeError(w, http.StatusNotFound, "file not found")
|
||||
return nil, false
|
||||
}
|
||||
if err != nil {
|
||||
writeInternalError(w, r, err, "failed to get file")
|
||||
return nil, false
|
||||
}
|
||||
return node, true
|
||||
}
|
||||
|
||||
func (h *NodeHandler) publicNode(w http.ResponseWriter, r *http.Request) (*model.Node, bool) {
|
||||
node, err := h.repo.GetByPublicToken(r.Context(), chi.URLParam(r, "token"))
|
||||
if errors.Is(err, repository.ErrNotFound) {
|
||||
writeError(w, http.StatusNotFound, "public link not found")
|
||||
return nil, false
|
||||
}
|
||||
if err != nil {
|
||||
writeInternalError(w, r, err, "failed to open public link")
|
||||
return nil, false
|
||||
}
|
||||
return node, true
|
||||
}
|
||||
|
||||
func (h *NodeHandler) streamNode(w http.ResponseWriter, r *http.Request, node *model.Node) {
|
||||
if node.NodeType == model.NodeTypeFolder || node.StorageKey == nil {
|
||||
writeError(w, http.StatusBadRequest, "node is not downloadable")
|
||||
return
|
||||
}
|
||||
info, err := h.store.Stat(r.Context(), *node.StorageKey)
|
||||
if err != nil {
|
||||
writeInternalError(w, r, err, "failed to open file")
|
||||
return
|
||||
}
|
||||
start, end, hasRange := storage.ParseRange(r.Header.Get("Range"), info.Size)
|
||||
body, info, err := h.store.GetObject(r.Context(), *node.StorageKey, start, end)
|
||||
if err != nil {
|
||||
writeInternalError(w, r, err, "failed to stream file")
|
||||
return
|
||||
}
|
||||
defer body.Close()
|
||||
filename := node.Title
|
||||
if node.OriginalFilename != nil && *node.OriginalFilename != "" {
|
||||
filename = *node.OriginalFilename
|
||||
}
|
||||
w.Header().Set("Content-Disposition", `inline; filename="`+strings.ReplaceAll(filename, `"`, "")+`"`)
|
||||
storage.WriteRangeResponse(w, info.ContentType, info.Size, start, end, hasRange)
|
||||
_, _ = io.Copy(w, body)
|
||||
}
|
||||
|
||||
func emptyToNil(v string) *string {
|
||||
v = strings.TrimSpace(v)
|
||||
if v == "" {
|
||||
return nil
|
||||
}
|
||||
return &v
|
||||
}
|
||||
|
||||
func subordinates(r *http.Request) []string {
|
||||
ids := csvHeader(r, "X-User-Subordinates")
|
||||
if len(ids) == 0 {
|
||||
ids = csvHeader(r, "X-User-Subordinate-Ids")
|
||||
}
|
||||
return ids
|
||||
}
|
||||
|
||||
func newToken() (string, error) {
|
||||
buf := make([]byte, 32)
|
||||
if _, err := rand.Read(buf); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return base64.RawURLEncoding.EncodeToString(buf), nil
|
||||
}
|
||||
54
internal/migrate/migrate.go
Normal file
54
internal/migrate/migrate.go
Normal file
@@ -0,0 +1,54 @@
|
||||
package migrate
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
func Run(ctx context.Context, pool *pgxpool.Pool, migrationsDir string) error {
|
||||
_, err := pool.Exec(ctx, `
|
||||
CREATE TABLE IF NOT EXISTS schema_migrations (
|
||||
version VARCHAR(255) PRIMARY KEY,
|
||||
applied_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
)
|
||||
`)
|
||||
if err != nil {
|
||||
return fmt.Errorf("create migrations table: %w", err)
|
||||
}
|
||||
|
||||
files, err := filepath.Glob(filepath.Join(migrationsDir, "*.up.sql"))
|
||||
if err != nil {
|
||||
return fmt.Errorf("glob migrations: %w", err)
|
||||
}
|
||||
sort.Strings(files)
|
||||
|
||||
for _, f := range files {
|
||||
version := strings.TrimSuffix(filepath.Base(f), ".up.sql")
|
||||
var exists bool
|
||||
if err := pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM schema_migrations WHERE version = $1)`, version).Scan(&exists); err != nil {
|
||||
return fmt.Errorf("check migration %s: %w", version, err)
|
||||
}
|
||||
if exists {
|
||||
continue
|
||||
}
|
||||
sql, err := os.ReadFile(f)
|
||||
if err != nil {
|
||||
return fmt.Errorf("read migration %s: %w", version, err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, string(sql)); err != nil {
|
||||
return fmt.Errorf("apply migration %s: %w", version, err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `INSERT INTO schema_migrations (version) VALUES ($1)`, version); err != nil {
|
||||
return fmt.Errorf("record migration %s: %w", version, err)
|
||||
}
|
||||
slog.Info("applied migration", "version", version)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
67
internal/model/model.go
Normal file
67
internal/model/model.go
Normal file
@@ -0,0 +1,67 @@
|
||||
package model
|
||||
|
||||
import "time"
|
||||
|
||||
const (
|
||||
NodeTypeFolder = "folder"
|
||||
NodeTypeFile = "file"
|
||||
NodeTypeGoogleSheet = "google_sheet"
|
||||
NodeTypeOfficeDocument = "office_document"
|
||||
|
||||
AccessView = "view"
|
||||
AccessEdit = "edit"
|
||||
)
|
||||
|
||||
type Node struct {
|
||||
ID string `json:"id"`
|
||||
ParentID *string `json:"parent_id,omitempty"`
|
||||
NodeType string `json:"node_type"`
|
||||
Title string `json:"title"`
|
||||
OwnerUserID string `json:"owner_user_id"`
|
||||
OwnerDepartmentID *string `json:"owner_department_id,omitempty"`
|
||||
CreatedBy string `json:"created_by"`
|
||||
UpdatedBy *string `json:"updated_by,omitempty"`
|
||||
StorageKey *string `json:"storage_key,omitempty"`
|
||||
OriginalFilename *string `json:"original_filename,omitempty"`
|
||||
MimeType *string `json:"mime_type,omitempty"`
|
||||
Extension *string `json:"extension,omitempty"`
|
||||
SizeBytes int64 `json:"size_bytes"`
|
||||
OfficeFormat *string `json:"office_format,omitempty"`
|
||||
ExternalURL *string `json:"external_url,omitempty"`
|
||||
Version int `json:"version"`
|
||||
EffectiveAccess string `json:"effective_access"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
DeletedAt *time.Time `json:"deleted_at,omitempty"`
|
||||
}
|
||||
|
||||
type Access struct {
|
||||
UserID string `json:"user_id"`
|
||||
AccessLevel string `json:"access_level"`
|
||||
GrantedBy string `json:"granted_by,omitempty"`
|
||||
CreatedAt time.Time `json:"created_at,omitempty"`
|
||||
}
|
||||
|
||||
type CreateFolderRequest struct {
|
||||
ParentID *string `json:"parent_id"`
|
||||
Title string `json:"title"`
|
||||
}
|
||||
|
||||
type UpdateNodeRequest struct {
|
||||
ParentID *string `json:"parent_id"`
|
||||
Title *string `json:"title"`
|
||||
}
|
||||
|
||||
type ReplaceAccessRequest struct {
|
||||
Access []Access `json:"access"`
|
||||
}
|
||||
|
||||
type PublicLinkRequest struct {
|
||||
ExpiresAt time.Time `json:"expires_at"`
|
||||
}
|
||||
|
||||
type PublicLinkResponse struct {
|
||||
ID string `json:"id"`
|
||||
URL string `json:"url"`
|
||||
ExpiresAt time.Time `json:"expires_at"`
|
||||
}
|
||||
283
internal/repository/node.go
Normal file
283
internal/repository/node.go
Normal file
@@ -0,0 +1,283 @@
|
||||
package repository
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"files-service/internal/model"
|
||||
)
|
||||
|
||||
var ErrNotFound = errors.New("not found")
|
||||
|
||||
type NodeRepository struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewNodeRepository(pool *pgxpool.Pool) *NodeRepository {
|
||||
return &NodeRepository{pool: pool}
|
||||
}
|
||||
|
||||
func scanNode(scan func(dest ...any) error) (*model.Node, error) {
|
||||
var n model.Node
|
||||
err := scan(
|
||||
&n.ID, &n.ParentID, &n.NodeType, &n.Title, &n.OwnerUserID, &n.OwnerDepartmentID,
|
||||
&n.CreatedBy, &n.UpdatedBy, &n.StorageKey, &n.OriginalFilename, &n.MimeType,
|
||||
&n.Extension, &n.SizeBytes, &n.OfficeFormat, &n.ExternalURL, &n.Version,
|
||||
&n.EffectiveAccess, &n.CreatedAt, &n.UpdatedAt, &n.DeletedAt,
|
||||
)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return nil, ErrNotFound
|
||||
}
|
||||
return &n, err
|
||||
}
|
||||
|
||||
func (r *NodeRepository) List(ctx context.Context, userID string, subordinateIDs []string, scope string, parentID *string) ([]model.Node, error) {
|
||||
args := []any{userID, subordinateIDs}
|
||||
where := []string{"n.deleted_at IS NULL"}
|
||||
if parentID == nil || *parentID == "" {
|
||||
where = append(where, "n.parent_id IS NULL")
|
||||
} else {
|
||||
args = append(args, *parentID)
|
||||
where = append(where, fmt.Sprintf("n.parent_id = $%d", len(args)))
|
||||
}
|
||||
|
||||
switch scope {
|
||||
case "shared":
|
||||
where = append(where, `n.owner_user_id <> $1 AND (
|
||||
has_node_access(n.id, $1)
|
||||
OR n.owner_user_id::text = ANY($2::text[])
|
||||
)`)
|
||||
default:
|
||||
where = append(where, "n.owner_user_id = $1")
|
||||
}
|
||||
|
||||
query := `
|
||||
SELECT n.id, n.parent_id, n.node_type, n.title, n.owner_user_id, n.owner_department_id,
|
||||
n.created_by, n.updated_by, n.storage_key, n.original_filename, n.mime_type,
|
||||
n.extension, n.size_bytes, n.office_format, n.external_url, n.version,
|
||||
effective_node_access(n.id, $1, $2::text[]),
|
||||
n.created_at, n.updated_at, n.deleted_at
|
||||
FROM files_nodes n
|
||||
WHERE ` + strings.Join(where, " AND ") + `
|
||||
ORDER BY CASE WHEN n.node_type = 'folder' THEN 0 ELSE 1 END, lower(n.title), n.created_at DESC`
|
||||
|
||||
rows, err := r.pool.Query(ctx, query, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
out := make([]model.Node, 0)
|
||||
for rows.Next() {
|
||||
n, err := scanNode(rows.Scan)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, *n)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
func (r *NodeRepository) GetForUser(ctx context.Context, id, userID string, subordinateIDs []string) (*model.Node, error) {
|
||||
return scanNode(r.pool.QueryRow(ctx, `
|
||||
SELECT n.id, n.parent_id, n.node_type, n.title, n.owner_user_id, n.owner_department_id,
|
||||
n.created_by, n.updated_by, n.storage_key, n.original_filename, n.mime_type,
|
||||
n.extension, n.size_bytes, n.office_format, n.external_url, n.version,
|
||||
effective_node_access(n.id, $2, $3::text[]),
|
||||
n.created_at, n.updated_at, n.deleted_at
|
||||
FROM files_nodes n
|
||||
WHERE n.id = $1
|
||||
AND n.deleted_at IS NULL
|
||||
AND effective_node_access(n.id, $2, $3::text[]) <> ''
|
||||
`, id, userID, subordinateIDs).Scan)
|
||||
}
|
||||
|
||||
func (r *NodeRepository) CreateFolder(ctx context.Context, title string, parentID *string, ownerID string) (*model.Node, error) {
|
||||
return scanNode(r.pool.QueryRow(ctx, `
|
||||
INSERT INTO files_nodes (parent_id, node_type, title, owner_user_id, created_by)
|
||||
VALUES ($1, 'folder', $2, $3, $3)
|
||||
RETURNING id, parent_id, node_type, title, owner_user_id, owner_department_id,
|
||||
created_by, updated_by, storage_key, original_filename, mime_type,
|
||||
extension, size_bytes, office_format, external_url, version,
|
||||
'edit' AS effective_access, created_at, updated_at, deleted_at
|
||||
`, parentID, title, ownerID).Scan)
|
||||
}
|
||||
|
||||
func (r *NodeRepository) CreateFile(ctx context.Context, n *model.Node) (*model.Node, error) {
|
||||
return scanNode(r.pool.QueryRow(ctx, `
|
||||
INSERT INTO files_nodes
|
||||
(parent_id, node_type, title, owner_user_id, created_by, storage_key,
|
||||
original_filename, mime_type, extension, size_bytes)
|
||||
VALUES ($1, 'file', $2, $3, $3, $4, $5, $6, $7, $8)
|
||||
RETURNING id, parent_id, node_type, title, owner_user_id, owner_department_id,
|
||||
created_by, updated_by, storage_key, original_filename, mime_type,
|
||||
extension, size_bytes, office_format, external_url, version,
|
||||
'edit' AS effective_access, created_at, updated_at, deleted_at
|
||||
`, n.ParentID, n.Title, n.OwnerUserID, n.StorageKey, n.OriginalFilename, n.MimeType, n.Extension, n.SizeBytes).Scan)
|
||||
}
|
||||
|
||||
func (r *NodeRepository) Update(ctx context.Context, id, actorID string, req model.UpdateNodeRequest) (*model.Node, error) {
|
||||
return scanNode(r.pool.QueryRow(ctx, `
|
||||
UPDATE files_nodes
|
||||
SET title = COALESCE($3, title),
|
||||
parent_id = COALESCE($4, parent_id),
|
||||
updated_by = $2,
|
||||
updated_at = now(),
|
||||
version = version + 1
|
||||
WHERE id = $1
|
||||
AND deleted_at IS NULL
|
||||
AND effective_node_access(id, $2, '{}'::text[]) = 'edit'
|
||||
RETURNING id, parent_id, node_type, title, owner_user_id, owner_department_id,
|
||||
created_by, updated_by, storage_key, original_filename, mime_type,
|
||||
extension, size_bytes, office_format, external_url, version,
|
||||
'edit' AS effective_access, created_at, updated_at, deleted_at
|
||||
`, id, actorID, req.Title, req.ParentID).Scan)
|
||||
}
|
||||
|
||||
func (r *NodeRepository) SoftDelete(ctx context.Context, id, actorID string) error {
|
||||
tag, err := r.pool.Exec(ctx, `
|
||||
WITH RECURSIVE subtree AS (
|
||||
SELECT id FROM files_nodes WHERE id = $1 AND deleted_at IS NULL
|
||||
UNION ALL
|
||||
SELECT c.id FROM files_nodes c JOIN subtree s ON c.parent_id = s.id WHERE c.deleted_at IS NULL
|
||||
)
|
||||
UPDATE files_nodes
|
||||
SET deleted_at = now(), updated_by = $2, updated_at = now()
|
||||
WHERE id IN (SELECT id FROM subtree)
|
||||
AND effective_node_access($1, $2, '{}'::text[]) = 'edit'
|
||||
`, id, actorID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if tag.RowsAffected() == 0 {
|
||||
return ErrNotFound
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *NodeRepository) ListAccess(ctx context.Context, nodeID string) ([]model.Access, error) {
|
||||
rows, err := r.pool.Query(ctx, `
|
||||
SELECT user_id, access_level, granted_by, created_at
|
||||
FROM files_access
|
||||
WHERE node_id = $1
|
||||
ORDER BY created_at DESC
|
||||
`, nodeID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
out := make([]model.Access, 0)
|
||||
for rows.Next() {
|
||||
var a model.Access
|
||||
if err := rows.Scan(&a.UserID, &a.AccessLevel, &a.GrantedBy, &a.CreatedAt); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, a)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
func (r *NodeRepository) ReplaceAccess(ctx context.Context, nodeID, actorID string, access []model.Access) error {
|
||||
tx, err := r.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
|
||||
var canEdit bool
|
||||
if err := tx.QueryRow(ctx, `SELECT effective_node_access($1, $2, '{}'::text[]) = 'edit'`, nodeID, actorID).Scan(&canEdit); err != nil {
|
||||
return err
|
||||
}
|
||||
if !canEdit {
|
||||
return ErrNotFound
|
||||
}
|
||||
if _, err := tx.Exec(ctx, `DELETE FROM files_access WHERE node_id = $1`, nodeID); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, a := range normalizeAccess(access) {
|
||||
if _, err := tx.Exec(ctx, `
|
||||
INSERT INTO files_access (node_id, user_id, access_level, granted_by)
|
||||
VALUES ($1, $2, $3, $4)
|
||||
ON CONFLICT (node_id, user_id)
|
||||
DO UPDATE SET access_level = EXCLUDED.access_level, granted_by = EXCLUDED.granted_by
|
||||
`, nodeID, a.UserID, a.AccessLevel, actorID); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return tx.Commit(ctx)
|
||||
}
|
||||
|
||||
func normalizeAccess(access []model.Access) []model.Access {
|
||||
seen := map[string]model.Access{}
|
||||
for _, a := range access {
|
||||
if a.UserID == "" {
|
||||
continue
|
||||
}
|
||||
if a.AccessLevel != model.AccessEdit {
|
||||
a.AccessLevel = model.AccessView
|
||||
}
|
||||
if prev, ok := seen[a.UserID]; ok && prev.AccessLevel == model.AccessEdit {
|
||||
continue
|
||||
}
|
||||
seen[a.UserID] = a
|
||||
}
|
||||
out := make([]model.Access, 0, len(seen))
|
||||
for _, a := range seen {
|
||||
out = append(out, a)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func (r *NodeRepository) CreatePublicLink(ctx context.Context, nodeID, actorID, token string, expiresAt time.Time) (string, error) {
|
||||
hash := TokenHash(token)
|
||||
var id string
|
||||
err := r.pool.QueryRow(ctx, `
|
||||
INSERT INTO files_public_links (node_id, token_hash, expires_at, created_by)
|
||||
SELECT $1, $2, $3, $4
|
||||
WHERE effective_node_access($1, $4, '{}'::text[]) = 'edit'
|
||||
RETURNING id
|
||||
`, nodeID, hash, expiresAt, actorID).Scan(&id)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return "", ErrNotFound
|
||||
}
|
||||
return id, err
|
||||
}
|
||||
|
||||
func (r *NodeRepository) GetByPublicToken(ctx context.Context, token string) (*model.Node, error) {
|
||||
return scanNode(r.pool.QueryRow(ctx, `
|
||||
SELECT n.id, n.parent_id, n.node_type, n.title, n.owner_user_id, n.owner_department_id,
|
||||
n.created_by, n.updated_by, n.storage_key, n.original_filename, n.mime_type,
|
||||
n.extension, n.size_bytes, n.office_format, n.external_url, n.version,
|
||||
'view' AS effective_access, n.created_at, n.updated_at, n.deleted_at
|
||||
FROM files_public_links l
|
||||
JOIN files_nodes n ON n.id = l.node_id
|
||||
WHERE l.token_hash = $1
|
||||
AND l.revoked_at IS NULL
|
||||
AND l.expires_at > now()
|
||||
AND n.deleted_at IS NULL
|
||||
`, TokenHash(token)).Scan)
|
||||
}
|
||||
|
||||
func (r *NodeRepository) Audit(ctx context.Context, actorID, action, entityType, entityID string, meta string) {
|
||||
if meta == "" {
|
||||
meta = "{}"
|
||||
}
|
||||
_, _ = r.pool.Exec(ctx, `
|
||||
INSERT INTO files_audit_events (actor_user_id, action, entity_type, entity_id, meta)
|
||||
VALUES (NULLIF($1, '')::uuid, $2, $3, NULLIF($4, '')::uuid, $5::jsonb)
|
||||
`, actorID, action, entityType, entityID, meta)
|
||||
}
|
||||
|
||||
func TokenHash(token string) string {
|
||||
sum := sha256.Sum256([]byte(token))
|
||||
return hex.EncodeToString(sum[:])
|
||||
}
|
||||
183
internal/storage/minio.go
Normal file
183
internal/storage/minio.go
Normal file
@@ -0,0 +1,183 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"mime"
|
||||
"net/http"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/minio/minio-go/v7"
|
||||
"github.com/minio/minio-go/v7/pkg/credentials"
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
Endpoint string
|
||||
AccessKey string
|
||||
SecretKey string
|
||||
Bucket string
|
||||
UseSSL bool
|
||||
}
|
||||
|
||||
type Storage struct {
|
||||
cfg Config
|
||||
client *minio.Client
|
||||
}
|
||||
|
||||
type ObjectInfo struct {
|
||||
Size int64
|
||||
ContentType string
|
||||
}
|
||||
|
||||
func New(cfg Config) (*Storage, error) {
|
||||
if cfg.Endpoint == "" || cfg.AccessKey == "" || cfg.SecretKey == "" {
|
||||
return &Storage{cfg: cfg}, nil
|
||||
}
|
||||
cli, err := minio.New(cfg.Endpoint, &minio.Options{
|
||||
Creds: credentials.NewStaticV4(cfg.AccessKey, cfg.SecretKey, ""),
|
||||
Secure: cfg.UseSSL,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("init minio client: %w", err)
|
||||
}
|
||||
return &Storage{cfg: cfg, client: cli}, nil
|
||||
}
|
||||
|
||||
func (s *Storage) Configured() bool {
|
||||
return s.client != nil && s.cfg.Bucket != ""
|
||||
}
|
||||
|
||||
func (s *Storage) EnsureBucket(ctx context.Context) error {
|
||||
if !s.Configured() {
|
||||
return nil
|
||||
}
|
||||
exists, err := s.client.BucketExists(ctx, s.cfg.Bucket)
|
||||
if err != nil {
|
||||
return fmt.Errorf("check bucket: %w", err)
|
||||
}
|
||||
if exists {
|
||||
return nil
|
||||
}
|
||||
return s.client.MakeBucket(ctx, s.cfg.Bucket, minio.MakeBucketOptions{})
|
||||
}
|
||||
|
||||
func GenerateKey(ownerID, filename string) string {
|
||||
ext := strings.ToLower(strings.TrimPrefix(filepath.Ext(filename), "."))
|
||||
if !AllowedExtension(ext) {
|
||||
ext = "bin"
|
||||
}
|
||||
return fmt.Sprintf("%s/%s.%s", ownerID, uuid.NewString(), ext)
|
||||
}
|
||||
|
||||
func AllowedExtension(ext string) bool {
|
||||
switch strings.ToLower(ext) {
|
||||
case "doc", "docx", "xls", "xlsx", "xlsm", "ppt", "pptx", "ods", "odt", "odp",
|
||||
"pdf", "png", "jpg", "jpeg", "webp", "gif", "mp4", "webm", "mov", "m4v", "mp3", "wav", "ogg":
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func GuessContentType(filename, clientType string) string {
|
||||
if clientType != "" && clientType != "application/octet-stream" {
|
||||
return clientType
|
||||
}
|
||||
if ext := filepath.Ext(filename); ext != "" {
|
||||
if v := mime.TypeByExtension(ext); v != "" {
|
||||
return v
|
||||
}
|
||||
}
|
||||
return "application/octet-stream"
|
||||
}
|
||||
|
||||
func (s *Storage) PutObject(ctx context.Context, key string, body io.Reader, size int64, contentType string) error {
|
||||
if !s.Configured() {
|
||||
return errors.New("storage not configured")
|
||||
}
|
||||
_, err := s.client.PutObject(ctx, s.cfg.Bucket, key, body, size, minio.PutObjectOptions{
|
||||
ContentType: contentType,
|
||||
})
|
||||
return err
|
||||
}
|
||||
|
||||
func (s *Storage) Stat(ctx context.Context, key string) (*ObjectInfo, error) {
|
||||
if !s.Configured() {
|
||||
return nil, errors.New("storage not configured")
|
||||
}
|
||||
info, err := s.client.StatObject(ctx, s.cfg.Bucket, key, minio.StatObjectOptions{})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &ObjectInfo{Size: info.Size, ContentType: info.ContentType}, nil
|
||||
}
|
||||
|
||||
func (s *Storage) GetObject(ctx context.Context, key string, rangeStart, rangeEnd int64) (io.ReadCloser, *ObjectInfo, error) {
|
||||
if !s.Configured() {
|
||||
return nil, nil, errors.New("storage not configured")
|
||||
}
|
||||
opts := minio.GetObjectOptions{}
|
||||
if rangeStart > 0 || rangeEnd > 0 {
|
||||
_ = opts.SetRange(rangeStart, rangeEnd)
|
||||
}
|
||||
obj, err := s.client.GetObject(ctx, s.cfg.Bucket, key, opts)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
info, err := obj.Stat()
|
||||
if err != nil {
|
||||
_ = obj.Close()
|
||||
return nil, nil, err
|
||||
}
|
||||
return obj, &ObjectInfo{Size: info.Size, ContentType: info.ContentType}, nil
|
||||
}
|
||||
|
||||
func ParseRange(header string, totalSize int64) (start, end int64, ok bool) {
|
||||
if !strings.HasPrefix(header, "bytes=") {
|
||||
return 0, 0, false
|
||||
}
|
||||
parts := strings.SplitN(strings.TrimPrefix(header, "bytes="), "-", 2)
|
||||
if len(parts) != 2 {
|
||||
return 0, 0, false
|
||||
}
|
||||
if parts[0] == "" {
|
||||
n, err := strconv.ParseInt(parts[1], 10, 64)
|
||||
if err != nil || n <= 0 || n > totalSize {
|
||||
return 0, 0, false
|
||||
}
|
||||
return totalSize - n, totalSize - 1, true
|
||||
}
|
||||
start, err := strconv.ParseInt(parts[0], 10, 64)
|
||||
if err != nil || start < 0 || start >= totalSize {
|
||||
return 0, 0, false
|
||||
}
|
||||
if parts[1] == "" {
|
||||
return start, totalSize - 1, true
|
||||
}
|
||||
end, err = strconv.ParseInt(parts[1], 10, 64)
|
||||
if err != nil || end < start {
|
||||
return 0, 0, false
|
||||
}
|
||||
if end >= totalSize {
|
||||
end = totalSize - 1
|
||||
}
|
||||
return start, end, true
|
||||
}
|
||||
|
||||
func WriteRangeResponse(w http.ResponseWriter, contentType string, totalSize, start, end int64, hasRange bool) {
|
||||
w.Header().Set("Content-Type", contentType)
|
||||
w.Header().Set("Accept-Ranges", "bytes")
|
||||
if hasRange {
|
||||
w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, end, totalSize))
|
||||
w.Header().Set("Content-Length", strconv.FormatInt(end-start+1, 10))
|
||||
w.WriteHeader(http.StatusPartialContent)
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Length", strconv.FormatInt(totalSize, 10))
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}
|
||||
Reference in New Issue
Block a user