jfraeysd/fetch_ml
0
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

feat(api): add structured error package and refactor handlers

Browse source 
New error handling:
- Add internal/api/errors/errors.go with structured API error types
- Standardize error codes across all API endpoints
- Add user-facing error messages vs internal error details separation

Handler improvements:
- jupyter/handlers.go: better workspace lifecycle and error handling
- plugins/handlers.go: plugin management with validation
- groups/handlers.go: group CRUD with capability metadata
- jobs/handlers.go: job submission and monitoring improvements
- datasets/handlers.go: dataset upload/download with progress
- validate/handlers.go: manifest validation with detailed errors
- audit/handlers.go: audit log querying with filters

Server configuration:
- server_config.go: refined config loading with validation
- server_gen.go: improved code generation for OpenAPI specs
This commit is contained in:
Jeremie Fraeys 2026-03-12 12:04:46 -04:00
parent 37c4d4e9c7
commit c18a8619fe
No known key found for this signature in database
10 changed files with 393 additions and 235 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
internal/api/audit/handlers.go
View file

@ -7,6 +7,7 @@ import (
"strconv"
"time"
"github.com/jfraeys/fetch_ml/internal/api/errors"
"github.com/jfraeys/fetch_ml/internal/auth"
"github.com/jfraeys/fetch_ml/internal/logging"
)
@ -74,7 +75,7 @@ func NewHandler(logger *logging.Logger, store AuditStore) *Handler {
func (h *Handler) GetV1AuditEvents(w http.ResponseWriter, r *http.Request) {
user := auth.GetUserFromContext(r.Context())
if !h.checkPermission(user, "audit:read") {
http.Error(w, `{"error":"Insufficient permissions","code":"FORBIDDEN"}`, http.StatusForbidden)
errors.WriteHTTPError(w, http.StatusForbidden, errors.CodePermissionDenied, "Insufficient permissions", "")
return
}
@ -104,7 +105,7 @@ func (h *Handler) GetV1AuditEvents(w http.ResponseWriter, r *http.Request) {
if h.store != nil {
events, total, err := h.store.QueryEvents(from, to, eventType, userID, limit, offset)
if err != nil {
http.Error(w, `{"error":"Failed to query events","code":"INTERNAL_ERROR"}`, http.StatusInternalServerError)
errors.WriteHTTPError(w, http.StatusInternalServerError, errors.CodeUnknownError, "Failed to query events", err.Error())
return
}
@ -140,7 +141,7 @@ func (h *Handler) GetV1AuditEvents(w http.ResponseWriter, r *http.Request) {
func (h *Handler) PostV1AuditVerify(w http.ResponseWriter, r *http.Request) {
user := auth.GetUserFromContext(r.Context())
if !h.checkPermission(user, "audit:verify") {
http.Error(w, `{"error":"Insufficient permissions","code":"FORBIDDEN"}`, http.StatusForbidden)
errors.WriteHTTPError(w, http.StatusForbidden, errors.CodePermissionDenied, "Insufficient permissions", "")
return
}
@ -163,7 +164,7 @@ func (h *Handler) PostV1AuditVerify(w http.ResponseWriter, r *http.Request) {
func (h *Handler) GetV1AuditChainRoot(w http.ResponseWriter, r *http.Request) {
user := auth.GetUserFromContext(r.Context())
if !h.checkPermission(user, "audit:read") {
http.Error(w, `{"error":"Insufficient permissions","code":"FORBIDDEN"}`, http.StatusForbidden)
errors.WriteHTTPError(w, http.StatusForbidden, errors.CodePermissionDenied, "Insufficient permissions", "")
return
}

43
internal/api/datasets/handlers.go
View file

@ -8,6 +8,7 @@ import (
"time"
"github.com/gorilla/websocket"
"github.com/jfraeys/fetch_ml/internal/api/errors"
"github.com/jfraeys/fetch_ml/internal/auth"
"github.com/jfraeys/fetch_ml/internal/logging"
"github.com/jfraeys/fetch_ml/internal/storage"
@ -33,27 +34,14 @@ func NewHandler(
}
}
// Error codes
const (
ErrorCodeInvalidRequest = 0x01
ErrorCodeAuthenticationFailed = 0x02
ErrorCodePermissionDenied = 0x03
ErrorCodeResourceNotFound = 0x04
)
// sendErrorPacket sends an error response packet to the client
func sendErrorPacket(conn *websocket.Conn, message string) error {
err := map[string]any{
"error": true,
"code": ErrorCodeInvalidRequest,
"message": message,
}
return conn.WriteJSON(err)
func sendErrorPacket(conn *websocket.Conn, code string, message string) error {
return errors.SendErrorPacket(conn, code, message, "")
}
// sendSuccessPacket sends a success response packet
func (h *Handler) sendSuccessPacket(conn *websocket.Conn, data map[string]any) error {
return conn.WriteJSON(data)
return errors.SendSuccessPacket(conn, data)
}
// sendDataPacket sends a data response packet
@ -89,7 +77,7 @@ func (h *Handler) HandleDatasetRegister(
conn *websocket.Conn, payload []byte, user *auth.User,
) error {
if len(payload) < 16+1+2 {
return sendErrorPacket(conn, "register dataset payload too short")
return sendErrorPacket(conn, errors.CodeInvalidRequest, "register dataset payload too short")
}
offset := 16
@ -97,7 +85,7 @@ func (h *Handler) HandleDatasetRegister(
nameLen := int(payload[offset])
offset++
if nameLen <= 0 || len(payload) < offset+nameLen+2 {
return sendErrorPacket(conn, "invalid name length")
return sendErrorPacket(conn, errors.CodeInvalidRequest, "invalid name length")
}
name := string(payload[offset : offset+nameLen])
offset += nameLen
@ -105,7 +93,7 @@ func (h *Handler) HandleDatasetRegister(
pathLen := int(binary.BigEndian.Uint16(payload[offset : offset+2]))
offset += 2
if pathLen < 0 || len(payload) < offset+pathLen {
return sendErrorPacket(conn, "invalid path length")
return sendErrorPacket(conn, errors.CodeInvalidRequest, "invalid path length")
}
path := string(payload[offset : offset+pathLen])
@ -123,11 +111,10 @@ func (h *Handler) HandleDatasetRegister(
}
return h.sendSuccessPacket(conn, map[string]any{
"success": true,
"name": name,
"path": path,
"user": user.Name,
"time": time.Now().UTC(),
"name": name,
"path": path,
"user": user.Name,
"time": time.Now().UTC(),
})
}
@ -135,7 +122,7 @@ func (h *Handler) HandleDatasetRegister(
// Protocol: [api_key_hash:16][dataset_id_len:1][dataset_id:var]
func (h *Handler) HandleDatasetInfo(conn *websocket.Conn, payload []byte, user *auth.User) error {
if len(payload) < 16+1 {
return sendErrorPacket(conn, "dataset info payload too short")
return sendErrorPacket(conn, errors.CodeInvalidRequest, "dataset info payload too short")
}
offset := 16
@ -143,7 +130,7 @@ func (h *Handler) HandleDatasetInfo(conn *websocket.Conn, payload []byte, user *
datasetIDLen := int(payload[offset])
offset++
if datasetIDLen <= 0 || len(payload) < offset+datasetIDLen {
return sendErrorPacket(conn, "invalid dataset ID length")
return sendErrorPacket(conn, errors.CodeInvalidRequest, "invalid dataset ID length")
}
datasetID := string(payload[offset : offset+datasetIDLen])
@ -168,7 +155,7 @@ func (h *Handler) HandleDatasetInfo(conn *websocket.Conn, payload []byte, user *
// Protocol: [api_key_hash:16][query_len:2][query:var]
func (h *Handler) HandleDatasetSearch(conn *websocket.Conn, payload []byte, user *auth.User) error {
if len(payload) < 16+2 {
return sendErrorPacket(conn, "dataset search payload too short")
return sendErrorPacket(conn, errors.CodeInvalidRequest, "dataset search payload too short")
}
offset := 16
@ -176,7 +163,7 @@ func (h *Handler) HandleDatasetSearch(conn *websocket.Conn, payload []byte, user
queryLen := int(binary.BigEndian.Uint16(payload[offset : offset+2]))
offset += 2
if queryLen < 0 || len(payload) < offset+queryLen {
return sendErrorPacket(conn, "invalid query length")
return sendErrorPacket(conn, errors.CodeInvalidRequest, "invalid query length")
}
query := string(payload[offset : offset+queryLen])

129
internal/api/errors/errors.go Normal file
View file

@ -0,0 +1,129 @@
// Package errors provides centralized error handling for the API
package errors
import (
"encoding/json"
"fmt"
"net/http"
"github.com/gorilla/websocket"
)
// Error codes - centralized to ensure consistency across all API handlers
const (
CodeUnknownError = "UNKNOWN_ERROR"
CodeInvalidRequest = "INVALID_REQUEST"
CodeAuthenticationFailed = "AUTHENTICATION_FAILED"
CodePermissionDenied = "PERMISSION_DENIED"
CodeResourceNotFound = "RESOURCE_NOT_FOUND"
CodeResourceAlreadyExists = "RESOURCE_ALREADY_EXISTS"
CodeServerOverloaded = "SERVER_OVERLOADED"
CodeDatabaseError = "DATABASE_ERROR"
CodeNetworkError = "NETWORK_ERROR"
CodeStorageError = "STORAGE_ERROR"
CodeTimeout = "TIMEOUT"
CodeJobNotFound = "JOB_NOT_FOUND"
CodeJobAlreadyRunning = "JOB_ALREADY_RUNNING"
CodeJobFailedToStart = "JOB_FAILED_TO_START"
CodeJobExecutionFailed = "JOB_EXECUTION_FAILED"
CodeJobCancelled = "JOB_CANCELLED"
CodeOutOfMemory = "OUT_OF_MEMORY"
CodeDiskFull = "DISK_FULL"
CodeInvalidConfiguration = "INVALID_CONFIGURATION"
CodeServiceUnavailable = "SERVICE_UNAVAILABLE"
CodeBadRequest = "BAD_REQUEST"
CodeForbidden = "FORBIDDEN"
CodeNotFound = "NOT_FOUND"
)
// ErrorResponse represents a standardized error response
type ErrorResponse struct {
ErrorMsg string `json:"error"`
Code string `json:"code"`
Message string `json:"message,omitempty"`
Details string `json:"details,omitempty"`
}
// NewErrorResponse creates a new error response
func NewErrorResponse(code, message, details string) ErrorResponse {
return ErrorResponse{
ErrorMsg: "true",
Code: code,
Message: message,
Details: details,
}
}
// Error implements the error interface
func (e ErrorResponse) Error() string {
if e.Details != "" {
return fmt.Sprintf("[%s] %s: %s", e.Code, e.Message, e.Details)
}
return fmt.Sprintf("[%s] %s", e.Code, e.Message)
}
// WriteHTTPError writes an error response to an HTTP ResponseWriter
func WriteHTTPError(w http.ResponseWriter, statusCode int, errCode, message, details string) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(statusCode)
resp := NewErrorResponse(errCode, message, details)
_ = json.NewEncoder(w).Encode(resp)
}
// WriteHTTPErrorFromError writes an error response from an error
func WriteHTTPErrorFromError(w http.ResponseWriter, statusCode int, err error) {
if errResp, ok := err.(ErrorResponse); ok {
WriteHTTPError(w, statusCode, errResp.Code, errResp.Message, errResp.Details)
return
}
WriteHTTPError(w, statusCode, CodeUnknownError, err.Error(), "")
}
// SendErrorPacket sends an error packet over WebSocket
func SendErrorPacket(conn *websocket.Conn, code, message, details string) error {
resp := NewErrorResponse(code, message, details)
return conn.WriteJSON(resp)
}
// SendSuccessPacket sends a success packet over WebSocket
// Automatically adds "success": true to the response
func SendSuccessPacket(conn *websocket.Conn, data map[string]any) error {
response := make(map[string]any, len(data)+1)
response["success"] = true
for k, v := range data {
response[k] = v
}
return conn.WriteJSON(response)
}
// Common error responses as typed errors for use with errors.Is/errors.As
var (
ErrNotFound = NewErrorResponse(CodeNotFound, "resource not found", "")
ErrPermissionDenied = NewErrorResponse(CodePermissionDenied, "permission denied", "")
ErrInvalidRequest = NewErrorResponse(CodeInvalidRequest, "invalid request", "")
ErrServiceUnavailable = NewErrorResponse(CodeServiceUnavailable, "service unavailable", "")
ErrServerOverloaded = NewErrorResponse(CodeServerOverloaded, "server overloaded", "")
)
// HTTPStatusFromCode maps error codes to HTTP status codes
func HTTPStatusFromCode(code string) int {
switch code {
case CodeInvalidRequest, CodeBadRequest:
return http.StatusBadRequest
case CodeAuthenticationFailed:
return http.StatusUnauthorized
case CodePermissionDenied, CodeForbidden:
return http.StatusForbidden
case CodeResourceNotFound, CodeNotFound, CodeJobNotFound:
return http.StatusNotFound
case CodeResourceAlreadyExists:
return http.StatusConflict
case CodeServerOverloaded, CodeServiceUnavailable:
return http.StatusServiceUnavailable
case CodeTimeout:
return http.StatusRequestTimeout
default:
return http.StatusInternalServerError
}
}

125
internal/api/groups/handlers.go
View file

@ -9,6 +9,7 @@ import (
"time"
"github.com/gorilla/websocket"
"github.com/jfraeys/fetch_ml/internal/api/errors"
"github.com/jfraeys/fetch_ml/internal/auth"
"github.com/jfraeys/fetch_ml/internal/logging"
"github.com/jfraeys/fetch_ml/internal/storage"
@ -91,7 +92,7 @@ func (h *Handler) ListGroups(w http.ResponseWriter, r *http.Request) {
}
w.Header().Set("Content-Type", "application/json")
if err := json.NewEncoder(w).Encode(map[string]interface{}{
if err := json.NewEncoder(w).Encode(map[string]any{
"groups": groups,
"count": len(groups),
}); err != nil {
@ -165,7 +166,7 @@ func (h *Handler) ListInvitations(w http.ResponseWriter, r *http.Request) {
}
w.Header().Set("Content-Type", "application/json")
if err := json.NewEncoder(w).Encode(map[string]interface{}{
if err := json.NewEncoder(w).Encode(map[string]any{
"invitations": invitations,
"count": len(invitations),
}); err != nil {
@ -350,7 +351,7 @@ func (h *Handler) ListGroupTasks(w http.ResponseWriter, r *http.Request) {
}
w.Header().Set("Content-Type", "application/json")
if err := json.NewEncoder(w).Encode(map[string]interface{}{
if err := json.NewEncoder(w).Encode(map[string]any{
"tasks": tasks,
"next_cursor": nextCursor,
"count": len(tasks),
@ -360,26 +361,20 @@ func (h *Handler) ListGroupTasks(w http.ResponseWriter, r *http.Request) {
}
// sendErrorPacket sends an error response packet to the client
func (h *Handler) sendErrorPacket(conn *websocket.Conn, code byte, message, details string) error {
err := map[string]interface{}{
"error": true,
"code": code,
"message": message,
"details": details,
}
return conn.WriteJSON(err)
func (h *Handler) sendErrorPacket(conn *websocket.Conn, code string, message, details string) error {
return errors.SendErrorPacket(conn, code, message, details)
}
// sendSuccessPacket sends a success response packet
func (h *Handler) sendSuccessPacket(conn *websocket.Conn, data map[string]interface{}) error {
return conn.WriteJSON(data)
func (h *Handler) sendSuccessPacket(conn *websocket.Conn, data map[string]any) error {
return errors.SendSuccessPacket(conn, data)
}
// HandleCreateGroup handles WebSocket group creation.
// Protocol: [api_key_hash:16][name_len:1][name:var][desc_len:2][desc:var]
func (h *Handler) HandleCreateGroup(conn *websocket.Conn, payload []byte, user *auth.User) error {
if len(payload) < 16+1+2 {
return h.sendErrorPacket(conn, 0x01, "payload too short", "")
return h.sendErrorPacket(conn, errors.CodeInvalidRequest, "payload too short", "")
}
offset := 16
@ -387,7 +382,7 @@ func (h *Handler) HandleCreateGroup(conn *websocket.Conn, payload []byte, user *
nameLen := int(payload[offset])
offset++
if nameLen <= 0 || len(payload) < offset+nameLen+2 {
return h.sendErrorPacket(conn, 0x01, "invalid name length", "")
return h.sendErrorPacket(conn, errors.CodeInvalidRequest, "invalid name length", "")
}
name := string(payload[offset : offset+nameLen])
offset += nameLen
@ -395,30 +390,29 @@ func (h *Handler) HandleCreateGroup(conn *websocket.Conn, payload []byte, user *
descLen := int(binary.BigEndian.Uint16(payload[offset : offset+2]))
offset += 2
if descLen < 0 || len(payload) < offset+descLen {
return h.sendErrorPacket(conn, 0x01, "invalid description length", "")
return h.sendErrorPacket(conn, errors.CodeInvalidRequest, "invalid description length", "")
}
description := string(payload[offset : offset+descLen])
// Only admins can create groups
if !user.Admin {
return h.sendErrorPacket(conn, 0x03, "forbidden: admin required", "")
return h.sendErrorPacket(conn, errors.CodePermissionDenied, "forbidden: admin required", "")
}
if name == "" {
return h.sendErrorPacket(conn, 0x01, "name is required", "")
return h.sendErrorPacket(conn, errors.CodeInvalidRequest, "name is required", "")
}
group, err := h.db.CreateGroup(name, description, user.Name)
if err != nil {
h.logger.Error("failed to create group", "error", err)
return h.sendErrorPacket(conn, 0x11, "failed to create group", err.Error())
return h.sendErrorPacket(conn, errors.CodeDatabaseError, "failed to create group", err.Error())
}
h.logger.Info("group created", "group_id", group.ID, "name", group.Name, "created_by", user.Name)
return h.sendSuccessPacket(conn, map[string]interface{}{
"success": true,
"group": group,
return h.sendSuccessPacket(conn, map[string]any{
"group": group,
})
}
@ -428,13 +422,12 @@ func (h *Handler) HandleListGroups(conn *websocket.Conn, payload []byte, user *a
groups, err := h.db.ListGroupsForUser(user.Name)
if err != nil {
h.logger.Error("failed to list groups", "error", err)
return h.sendErrorPacket(conn, 0x11, "failed to list groups", err.Error())
return h.sendErrorPacket(conn, errors.CodeDatabaseError, "failed to list groups", err.Error())
}
return h.sendSuccessPacket(conn, map[string]interface{}{
"success": true,
"groups": groups,
"count": len(groups),
return h.sendSuccessPacket(conn, map[string]any{
"groups": groups,
"count": len(groups),
})
}
@ -442,7 +435,7 @@ func (h *Handler) HandleListGroups(conn *websocket.Conn, payload []byte, user *a
// Protocol: [api_key_hash:16][group_id_len:2][group_id:var][user_id_len:2][user_id:var]
func (h *Handler) HandleCreateInvitation(conn *websocket.Conn, payload []byte, user *auth.User) error {
if len(payload) < 16+2+2 {
return h.sendErrorPacket(conn, 0x01, "payload too short", "")
return h.sendErrorPacket(conn, errors.CodeInvalidRequest, "payload too short", "")
}
offset := 16
@ -450,7 +443,7 @@ func (h *Handler) HandleCreateInvitation(conn *websocket.Conn, payload []byte, u
groupIDLen := int(binary.BigEndian.Uint16(payload[offset : offset+2]))
offset += 2
if groupIDLen <= 0 || len(payload) < offset+groupIDLen+2 {
return h.sendErrorPacket(conn, 0x01, "invalid group_id length", "")
return h.sendErrorPacket(conn, errors.CodeInvalidRequest, "invalid group_id length", "")
}
groupID := string(payload[offset : offset+groupIDLen])
offset += groupIDLen
@ -458,26 +451,25 @@ func (h *Handler) HandleCreateInvitation(conn *websocket.Conn, payload []byte, u
userIDLen := int(binary.BigEndian.Uint16(payload[offset : offset+2]))
offset += 2
if userIDLen <= 0 || len(payload) < offset+userIDLen {
return h.sendErrorPacket(conn, 0x01, "invalid user_id length", "")
return h.sendErrorPacket(conn, errors.CodeInvalidRequest, "invalid user_id length", "")
}
invitedUserID := string(payload[offset : offset+userIDLen])
// Check if user is group admin
isAdmin, err := h.db.IsGroupAdmin(user.Name, groupID)
if err != nil || !isAdmin {
return h.sendErrorPacket(conn, 0x03, "forbidden: group admin required", "")
return h.sendErrorPacket(conn, errors.CodePermissionDenied, "forbidden: group admin required", "")
}
invitation, err := h.db.CreateGroupInvitation(groupID, invitedUserID, user.Name)
if err != nil {
h.logger.Error("failed to create invitation", "error", err)
return h.sendErrorPacket(conn, 0x11, "failed to create invitation", err.Error())
return h.sendErrorPacket(conn, errors.CodeDatabaseError, "failed to create invitation", err.Error())
}
h.logger.Info("invitation created", "invitation_id", invitation.ID, "group_id", groupID, "invited_user", invitedUserID)
return h.sendSuccessPacket(conn, map[string]interface{}{
"success": true,
return h.sendSuccessPacket(conn, map[string]any{
"invitation": invitation,
})
}
@ -488,11 +480,10 @@ func (h *Handler) HandleListInvitations(conn *websocket.Conn, payload []byte, us
invitations, err := h.db.ListPendingInvitationsForUser(user.Name)
if err != nil {
h.logger.Error("failed to list invitations", "error", err)
return h.sendErrorPacket(conn, 0x11, "failed to list invitations", err.Error())
return h.sendErrorPacket(conn, errors.CodeDatabaseError, "failed to list invitations", err.Error())
}
return h.sendSuccessPacket(conn, map[string]interface{}{
"success": true,
return h.sendSuccessPacket(conn, map[string]any{
"invitations": invitations,
"count": len(invitations),
})
@ -502,7 +493,7 @@ func (h *Handler) HandleListInvitations(conn *websocket.Conn, payload []byte, us
// Protocol: [api_key_hash:16][invitation_id_len:2][invitation_id:var]
func (h *Handler) HandleAcceptInvitation(conn *websocket.Conn, payload []byte, user *auth.User) error {
if len(payload) < 16+2 {
return h.sendErrorPacket(conn, 0x01, "payload too short", "")
return h.sendErrorPacket(conn, errors.CodeInvalidRequest, "payload too short", "")
}
offset := 16
@ -510,38 +501,37 @@ func (h *Handler) HandleAcceptInvitation(conn *websocket.Conn, payload []byte, u
invIDLen := int(binary.BigEndian.Uint16(payload[offset : offset+2]))
offset += 2
if invIDLen <= 0 || len(payload) < offset+invIDLen {
return h.sendErrorPacket(conn, 0x01, "invalid invitation_id length", "")
return h.sendErrorPacket(conn, errors.CodeInvalidRequest, "invalid invitation_id length", "")
}
invitationID := string(payload[offset : offset+invIDLen])
// Verify invitation belongs to this user and is pending
invitation, err := h.db.GetInvitation(invitationID)
if err != nil {
return h.sendErrorPacket(conn, 0x04, "invitation not found", "")
return h.sendErrorPacket(conn, errors.CodeResourceNotFound, "invitation not found", "")
}
if invitation.InvitedUserID != user.Name {
return h.sendErrorPacket(conn, 0x03, "forbidden", "")
return h.sendErrorPacket(conn, errors.CodePermissionDenied, "forbidden", "")
}
if invitation.Status != "pending" {
return h.sendErrorPacket(conn, 0x05, "invitation already processed", "")
return h.sendErrorPacket(conn, errors.CodeResourceAlreadyExists, "invitation already processed", "")
}
// Check if expired (7 days default)
if invitation.ExpiresAt != nil && time.Now().After(*invitation.ExpiresAt) {
return h.sendErrorPacket(conn, 0x14, "invitation expired", "")
return h.sendErrorPacket(conn, errors.CodeTimeout, "invitation expired", "")
}
if err := h.db.AcceptInvitation(invitationID, user.Name); err != nil {
h.logger.Error("failed to accept invitation", "error", err)
return h.sendErrorPacket(conn, 0x11, "failed to accept invitation", err.Error())
return h.sendErrorPacket(conn, errors.CodeDatabaseError, "failed to accept invitation", err.Error())
}
h.logger.Info("invitation accepted", "invitation_id", invitationID, "user", user.Name, "group_id", invitation.GroupID)
return h.sendSuccessPacket(conn, map[string]interface{}{
"success": true,
return h.sendSuccessPacket(conn, map[string]any{
"status": "accepted",
"group_id": invitation.GroupID,
})
@ -551,7 +541,7 @@ func (h *Handler) HandleAcceptInvitation(conn *websocket.Conn, payload []byte, u
// Protocol: [api_key_hash:16][invitation_id_len:2][invitation_id:var]
func (h *Handler) HandleDeclineInvitation(conn *websocket.Conn, payload []byte, user *auth.User) error {
if len(payload) < 16+2 {
return h.sendErrorPacket(conn, 0x01, "payload too short", "")
return h.sendErrorPacket(conn, errors.CodeInvalidRequest, "payload too short", "")
}
offset := 16
@ -559,29 +549,28 @@ func (h *Handler) HandleDeclineInvitation(conn *websocket.Conn, payload []byte,
invIDLen := int(binary.BigEndian.Uint16(payload[offset : offset+2]))
offset += 2
if invIDLen <= 0 || len(payload) < offset+invIDLen {
return h.sendErrorPacket(conn, 0x01, "invalid invitation_id length", "")
return h.sendErrorPacket(conn, errors.CodeInvalidRequest, "invalid invitation_id length", "")
}
invitationID := string(payload[offset : offset+invIDLen])
invitation, err := h.db.GetInvitation(invitationID)
if err != nil {
return h.sendErrorPacket(conn, 0x04, "invitation not found", "")
return h.sendErrorPacket(conn, errors.CodeResourceNotFound, "invitation not found", "")
}
if invitation.InvitedUserID != user.Name {
return h.sendErrorPacket(conn, 0x03, "forbidden", "")
return h.sendErrorPacket(conn, errors.CodePermissionDenied, "forbidden", "")
}
if err := h.db.DeclineInvitation(invitationID, user.Name); err != nil {
h.logger.Error("failed to decline invitation", "error", err)
return h.sendErrorPacket(conn, 0x11, "failed to decline invitation", err.Error())
return h.sendErrorPacket(conn, errors.CodeDatabaseError, "failed to decline invitation", err.Error())
}
h.logger.Info("invitation declined", "invitation_id", invitationID, "user", user.Name)
return h.sendSuccessPacket(conn, map[string]interface{}{
"success": true,
"status": "declined",
return h.sendSuccessPacket(conn, map[string]any{
"status": "declined",
})
}
@ -589,7 +578,7 @@ func (h *Handler) HandleDeclineInvitation(conn *websocket.Conn, payload []byte,
// Protocol: [api_key_hash:16][group_id_len:2][group_id:var][user_id_len:2][user_id:var]
func (h *Handler) HandleRemoveMember(conn *websocket.Conn, payload []byte, user *auth.User) error {
if len(payload) < 16+2+2 {
return h.sendErrorPacket(conn, 0x01, "payload too short", "")
return h.sendErrorPacket(conn, errors.CodeInvalidRequest, "payload too short", "")
}
offset := 16
@ -597,7 +586,7 @@ func (h *Handler) HandleRemoveMember(conn *websocket.Conn, payload []byte, user
groupIDLen := int(binary.BigEndian.Uint16(payload[offset : offset+2]))
offset += 2
if groupIDLen <= 0 || len(payload) < offset+groupIDLen+2 {
return h.sendErrorPacket(conn, 0x01, "invalid group_id length", "")
return h.sendErrorPacket(conn, errors.CodeInvalidRequest, "invalid group_id length", "")
}
groupID := string(payload[offset : offset+groupIDLen])
offset += groupIDLen
@ -605,30 +594,29 @@ func (h *Handler) HandleRemoveMember(conn *websocket.Conn, payload []byte, user
userIDLen := int(binary.BigEndian.Uint16(payload[offset : offset+2]))
offset += 2
if userIDLen <= 0 || len(payload) < offset+userIDLen {
return h.sendErrorPacket(conn, 0x01, "invalid user_id length", "")
return h.sendErrorPacket(conn, errors.CodeInvalidRequest, "invalid user_id length", "")
}
memberID := string(payload[offset : offset+userIDLen])
// Check if user is group admin
isAdmin, err := h.db.IsGroupAdmin(user.Name, groupID)
if err != nil || !isAdmin {
return h.sendErrorPacket(conn, 0x03, "forbidden: group admin required", "")
return h.sendErrorPacket(conn, errors.CodePermissionDenied, "forbidden: group admin required", "")
}
// Cannot remove yourself
if memberID == user.Name {
return h.sendErrorPacket(conn, 0x01, "cannot remove yourself; use leave group endpoint", "")
return h.sendErrorPacket(conn, errors.CodeInvalidRequest, "cannot remove yourself; use leave group endpoint", "")
}
if err := h.db.RemoveGroupMember(groupID, memberID); err != nil {
h.logger.Error("failed to remove member", "error", err)
return h.sendErrorPacket(conn, 0x11, "failed to remove member", err.Error())
return h.sendErrorPacket(conn, errors.CodeDatabaseError, "failed to remove member", err.Error())
}
h.logger.Info("member removed", "group_id", groupID, "member", memberID, "removed_by", user.Name)
return h.sendSuccessPacket(conn, map[string]interface{}{
"success": true,
return h.sendSuccessPacket(conn, map[string]any{
"message": "Member removed",
})
}
@ -637,7 +625,7 @@ func (h *Handler) HandleRemoveMember(conn *websocket.Conn, payload []byte, user
// Protocol: [api_key_hash:16][group_id_len:2][group_id:var][limit:1][cursor_len:2][cursor:var]
func (h *Handler) HandleListGroupTasks(conn *websocket.Conn, payload []byte, user *auth.User) error {
if len(payload) < 16+2+1+2 {
return h.sendErrorPacket(conn, 0x01, "payload too short", "")
return h.sendErrorPacket(conn, errors.CodeInvalidRequest, "payload too short", "")
}
offset := 16
@ -645,7 +633,7 @@ func (h *Handler) HandleListGroupTasks(conn *websocket.Conn, payload []byte, use
groupIDLen := int(binary.BigEndian.Uint16(payload[offset : offset+2]))
offset += 2
if groupIDLen <= 0 || len(payload) < offset+groupIDLen+1+2 {
return h.sendErrorPacket(conn, 0x01, "invalid group_id length", "")
return h.sendErrorPacket(conn, errors.CodeInvalidRequest, "invalid group_id length", "")
}
groupID := string(payload[offset : offset+groupIDLen])
offset += groupIDLen
@ -653,7 +641,7 @@ func (h *Handler) HandleListGroupTasks(conn *websocket.Conn, payload []byte, use
// Check if user is member of group
isMember, err := h.db.IsGroupMember(user.Name, groupID)
if err != nil || !isMember {
return h.sendErrorPacket(conn, 0x03, "forbidden: group membership required", "")
return h.sendErrorPacket(conn, errors.CodePermissionDenied, "forbidden: group membership required", "")
}
limit := int(payload[offset])
@ -667,7 +655,7 @@ func (h *Handler) HandleListGroupTasks(conn *websocket.Conn, payload []byte, use
var cursor string
if cursorLen > 0 {
if len(payload) < offset+cursorLen {
return h.sendErrorPacket(conn, 0x01, "invalid cursor length", "")
return h.sendErrorPacket(conn, errors.CodeInvalidRequest, "invalid cursor length", "")
}
cursor = string(payload[offset : offset+cursorLen])
}
@ -680,11 +668,10 @@ func (h *Handler) HandleListGroupTasks(conn *websocket.Conn, payload []byte, use
tasks, nextCursor, err := h.db.ListTasksForGroup(groupID, opts)
if err != nil {
h.logger.Error("failed to list group tasks", "error", err)
return h.sendErrorPacket(conn, 0x11, "failed to list tasks", err.Error())
return h.sendErrorPacket(conn, errors.CodeDatabaseError, "failed to list tasks", err.Error())
}
return h.sendSuccessPacket(conn, map[string]interface{}{
"success": true,
return h.sendSuccessPacket(conn, map[string]any{
"tasks": tasks,
"next_cursor": nextCursor,
"count": len(tasks),

74
internal/api/jobs/handlers.go
View file

@ -11,6 +11,7 @@ import (
"time"
"github.com/gorilla/websocket"
"github.com/jfraeys/fetch_ml/internal/api/errors"
"github.com/jfraeys/fetch_ml/internal/auth"
"github.com/jfraeys/fetch_ml/internal/container"
"github.com/jfraeys/fetch_ml/internal/experiment"
@ -45,17 +46,17 @@ func NewHandler(
}
}
// Error codes
// Error codes - using standardized error codes from errors package
const (
ErrorCodeUnknownError = 0x00
ErrorCodeInvalidRequest = 0x01
ErrorCodeAuthenticationFailed = 0x02
ErrorCodePermissionDenied = 0x03
ErrorCodeResourceNotFound = 0x04
ErrorCodeResourceAlreadyExists = 0x05
ErrorCodeInvalidConfiguration = 0x32
ErrorCodeJobNotFound = 0x20
ErrorCodeJobAlreadyRunning = 0x21
ErrorCodeUnknownError = errors.CodeUnknownError
ErrorCodeInvalidRequest = errors.CodeInvalidRequest
ErrorCodeAuthenticationFailed = errors.CodeAuthenticationFailed
ErrorCodePermissionDenied = errors.CodePermissionDenied
ErrorCodeResourceNotFound = errors.CodeResourceNotFound
ErrorCodeResourceAlreadyExists = errors.CodeResourceAlreadyExists
ErrorCodeInvalidConfiguration = errors.CodeInvalidConfiguration
ErrorCodeJobNotFound = errors.CodeJobNotFound
ErrorCodeJobAlreadyRunning = errors.CodeJobAlreadyRunning
)
// Permissions
@ -66,19 +67,13 @@ const (
)
// sendErrorPacket sends an error response packet to the client
func (h *Handler) sendErrorPacket(conn *websocket.Conn, code byte, message, details string) error {
err := map[string]any{
"error": true,
"code": code,
"message": message,
"details": details,
}
return conn.WriteJSON(err)
func (h *Handler) sendErrorPacket(conn *websocket.Conn, code string, message, details string) error {
return errors.SendErrorPacket(conn, code, message, details)
}
// sendSuccessPacket sends a success response packet
func (h *Handler) sendSuccessPacket(conn *websocket.Conn, data map[string]any) error {
return conn.WriteJSON(data)
return errors.SendSuccessPacket(conn, data)
}
// HandleAnnotateRun handles the annotate run WebSocket operation
@ -145,8 +140,7 @@ func (h *Handler) HandleAnnotateRun(conn *websocket.Conn, payload []byte, user *
h.logger.Info("annotating run", "job", jobName, "author", author, "dir", manifestDir)
return h.sendSuccessPacket(conn, map[string]interface{}{
"success": true,
return h.sendSuccessPacket(conn, map[string]any{
"job_name": jobName,
"timestamp": time.Now().UTC(),
"note": note,
@ -213,8 +207,7 @@ func (h *Handler) HandleSetRunNarrative(conn *websocket.Conn, payload []byte, us
h.logger.Info("setting run narrative", "job", jobName, "bucket", bucket)
return h.sendSuccessPacket(conn, map[string]interface{}{
"success": true,
return h.sendSuccessPacket(conn, map[string]any{
"job_name": jobName,
"narrative": patch,
})
@ -283,8 +276,7 @@ func (h *Handler) HandleSetRunPrivacy(conn *websocket.Conn, payload []byte, user
h.logger.Info("setting run privacy", "job", jobName, "bucket", bucket, "user", user.Name)
return h.sendSuccessPacket(conn, map[string]interface{}{
"success": true,
return h.sendSuccessPacket(conn, map[string]any{
"job_name": jobName,
"privacy": patch,
})
@ -305,8 +297,7 @@ func (h *Handler) HandleCancelJob(conn *websocket.Conn, jobName string, user *au
}
}
return h.sendSuccessPacket(conn, map[string]interface{}{
"success": true,
return h.sendSuccessPacket(conn, map[string]any{
"job_name": jobName,
"message": "Cancellation requested",
})
@ -316,10 +307,9 @@ func (h *Handler) HandleCancelJob(conn *websocket.Conn, jobName string, user *au
func (h *Handler) HandlePruneJobs(conn *websocket.Conn, pruneType byte, value int, user *auth.User) error {
h.logger.Info("pruning jobs", "type", pruneType, "value", value, "user", user.Name)
return h.sendSuccessPacket(conn, map[string]interface{}{
"success": true,
"pruned": 0,
"type": pruneType,
return h.sendSuccessPacket(conn, map[string]any{
"pruned": 0,
"type": pruneType,
})
}
@ -360,7 +350,7 @@ func (h *Handler) HandleListJobs(conn *websocket.Conn, user *auth.User) error {
jobName := entry.Name()
jobs = append(jobs, map[string]interface{}{
jobs = append(jobs, map[string]any{
"name": jobName,
"status": "unknown",
"bucket": bucket,
@ -368,10 +358,9 @@ func (h *Handler) HandleListJobs(conn *websocket.Conn, user *auth.User) error {
}
}
return h.sendSuccessPacket(conn, map[string]interface{}{
"success": true,
"jobs": jobs,
"count": len(jobs),
return h.sendSuccessPacket(conn, map[string]any{
"jobs": jobs,
"count": len(jobs),
})
}
@ -417,14 +406,14 @@ func (h *Handler) GetExperimentHistoryHTTP(w http.ResponseWriter, r *http.Reques
"details": "Configuration loaded",
},
},
"annotations": []map[string]interface{}{
"annotations": []map[string]any{
{
"author": "user",
"timestamp": time.Now().UTC(),
"note": "Initial run - baseline results",
},
},
"config_snapshot": map[string]interface{}{
"config_snapshot": map[string]any{
"learning_rate": 0.001,
"batch_size": 32,
"epochs": 100,
@ -476,7 +465,7 @@ func (h *Handler) ListAllJobsHTTP(w http.ResponseWriter, r *http.Request) {
}
jobName := entry.Name()
job := map[string]interface{}{
job := map[string]any{
"name": jobName,
"status": "unknown",
"bucket": bucket,
@ -495,10 +484,9 @@ func (h *Handler) ListAllJobsHTTP(w http.ResponseWriter, r *http.Request) {
}
response := map[string]any{
"success": true,
"jobs": jobs,
"count": len(jobs),
"view": map[bool]string{true: "team", false: "personal"}[allUsers],
"jobs": jobs,
"count": len(jobs),
"view": map[bool]string{true: "team", false: "personal"}[allUsers],
}
w.Header().Set("Content-Type", "application/json")

133
internal/api/jupyter/handlers.go
View file

@ -3,10 +3,12 @@ package jupyter
import (
"encoding/binary"
"encoding/json"
"net/http"
"time"
"github.com/gorilla/websocket"
"github.com/jfraeys/fetch_ml/internal/api/errors"
"github.com/jfraeys/fetch_ml/internal/auth"
"github.com/jfraeys/fetch_ml/internal/container"
"github.com/jfraeys/fetch_ml/internal/jupyter"
@ -33,13 +35,13 @@ func NewHandler(
}
}
// Error codes
// Error codes - using standardized error codes from errors package
const (
ErrorCodeInvalidRequest = 0x01
ErrorCodeAuthenticationFailed = 0x02
ErrorCodePermissionDenied = 0x03
ErrorCodeResourceNotFound = 0x04
ErrorCodeServiceUnavailable = 0x33
ErrorCodeInvalidRequest = errors.CodeInvalidRequest
ErrorCodeAuthenticationFailed = errors.CodeAuthenticationFailed
ErrorCodePermissionDenied = errors.CodePermissionDenied
ErrorCodeResourceNotFound = errors.CodeResourceNotFound
ErrorCodeServiceUnavailable = errors.CodeServiceUnavailable
)
// Permissions
@ -49,19 +51,13 @@ const (
)
// sendErrorPacket sends an error response packet to the client
func (h *Handler) sendErrorPacket(conn *websocket.Conn, code byte, message, details string) error {
err := map[string]interface{}{
"error": true,
"code": code,
"message": message,
"details": details,
}
return conn.WriteJSON(err)
func (h *Handler) sendErrorPacket(conn *websocket.Conn, code string, message, details string) error {
return errors.SendErrorPacket(conn, code, message, details)
}
// sendSuccessPacket sends a success response packet
func (h *Handler) sendSuccessPacket(conn *websocket.Conn, data map[string]interface{}) error {
return conn.WriteJSON(data)
func (h *Handler) sendSuccessPacket(conn *websocket.Conn, data map[string]any) error {
return errors.SendSuccessPacket(conn, data)
}
// HandleStartJupyter handles starting a Jupyter service
@ -98,8 +94,7 @@ func (h *Handler) HandleStartJupyter(conn *websocket.Conn, payload []byte, user
return h.sendErrorPacket(conn, ErrorCodeServiceUnavailable, "Jupyter service manager not available", "")
}
return h.sendSuccessPacket(conn, map[string]interface{}{
"success": true,
return h.sendSuccessPacket(conn, map[string]any{
"workspace": workspace,
"timestamp": time.Now().UTC(),
})
@ -127,8 +122,7 @@ func (h *Handler) HandleStopJupyter(conn *websocket.Conn, payload []byte, user *
return h.sendErrorPacket(conn, ErrorCodeServiceUnavailable, "Jupyter service manager not available", "")
}
return h.sendSuccessPacket(conn, map[string]interface{}{
"success": true,
return h.sendSuccessPacket(conn, map[string]any{
"service_id": serviceID,
"timestamp": time.Now().UTC(),
})
@ -140,17 +134,15 @@ func (h *Handler) HandleListJupyter(conn *websocket.Conn, payload []byte, user *
h.logger.Info("listing jupyter services", "user", user.Name)
if h.jupyterMgr == nil {
return h.sendSuccessPacket(conn, map[string]interface{}{
"success": true,
"services": []interface{}{},
return h.sendSuccessPacket(conn, map[string]any{
"services": []any{},
"count": 0,
})
}
services := h.jupyterMgr.ListServices()
return h.sendSuccessPacket(conn, map[string]interface{}{
"success": true,
return h.sendSuccessPacket(conn, map[string]any{
"services": services,
"count": len(services),
})
@ -174,10 +166,9 @@ func (h *Handler) HandleListJupyterPackages(conn *websocket.Conn, payload []byte
h.logger.Info("listing jupyter packages", "service", serviceName, "user", user.Name)
return h.sendSuccessPacket(conn, map[string]interface{}{
"success": true,
return h.sendSuccessPacket(conn, map[string]any{
"service_name": serviceName,
"packages": []interface{}{},
"packages": []any{},
"count": 0,
})
}
@ -207,8 +198,7 @@ func (h *Handler) HandleRemoveJupyter(conn *websocket.Conn, payload []byte, user
return h.sendErrorPacket(conn, ErrorCodeServiceUnavailable, "Jupyter service manager not available", "")
}
return h.sendSuccessPacket(conn, map[string]interface{}{
"success": true,
return h.sendSuccessPacket(conn, map[string]any{
"service_id": serviceID,
"purged": purge,
})
@ -236,8 +226,7 @@ func (h *Handler) HandleRestoreJupyter(conn *websocket.Conn, payload []byte, use
return h.sendErrorPacket(conn, ErrorCodeServiceUnavailable, "Jupyter service manager not available", "")
}
return h.sendSuccessPacket(conn, map[string]interface{}{
"success": true,
return h.sendSuccessPacket(conn, map[string]any{
"workspace": workspace,
"restored": true,
})
@ -247,14 +236,88 @@ func (h *Handler) HandleRestoreJupyter(conn *websocket.Conn, payload []byte, use
// ListServicesHTTP handles HTTP requests for listing Jupyter services
func (h *Handler) ListServicesHTTP(w http.ResponseWriter, r *http.Request) {
http.Error(w, "Not implemented", http.StatusNotImplemented)
if h.jupyterMgr == nil {
errors.WriteHTTPError(w, http.StatusServiceUnavailable, errors.CodeServiceUnavailable, "Jupyter service manager not available", "")
return
}
services := h.jupyterMgr.ListServices()
w.Header().Set("Content-Type", "application/json")
response := map[string]any{
"services": services,
"count": len(services),
}
if err := json.NewEncoder(w).Encode(response); err != nil {
h.logger.Warn("failed to encode services list", "error", err)
}
}
// StartServiceHTTP handles HTTP requests for starting Jupyter service
func (h *Handler) StartServiceHTTP(w http.ResponseWriter, r *http.Request) {
http.Error(w, "Not implemented", http.StatusNotImplemented)
if h.jupyterMgr == nil {
errors.WriteHTTPError(w, http.StatusServiceUnavailable, errors.CodeServiceUnavailable, "Jupyter service manager not available", "")
return
}
var req struct {
Workspace string `json:"workspace"`
Config map[string]any `json:"config,omitempty"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
errors.WriteHTTPError(w, http.StatusBadRequest, errors.CodeInvalidRequest, "Invalid request body", "")
return
}
if req.Workspace == "" {
errors.WriteHTTPError(w, http.StatusBadRequest, errors.CodeInvalidRequest, "Workspace name is required", "")
return
}
if err := container.ValidateJobName(req.Workspace); err != nil {
errors.WriteHTTPError(w, http.StatusBadRequest, errors.CodeInvalidRequest, "Invalid workspace name", err.Error())
return
}
h.logger.Info("starting jupyter service via HTTP", "workspace", req.Workspace)
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusCreated)
response := map[string]any{
"workspace": req.Workspace,
"timestamp": time.Now().UTC(),
}
if err := json.NewEncoder(w).Encode(response); err != nil {
h.logger.Warn("failed to encode start response", "error", err)
}
}
// StopServiceHTTP handles HTTP requests for stopping Jupyter service
func (h *Handler) StopServiceHTTP(w http.ResponseWriter, r *http.Request) {
http.Error(w, "Not mplementated", http.StatusNotImplemented)
if h.jupyterMgr == nil {
errors.WriteHTTPError(w, http.StatusServiceUnavailable, errors.CodeServiceUnavailable, "Jupyter service manager not available", "")
return
}
serviceID := r.PathValue("serviceId")
if serviceID == "" {
serviceID = r.URL.Query().Get("service_id")
}
if serviceID == "" {
errors.WriteHTTPError(w, http.StatusBadRequest, errors.CodeInvalidRequest, "Service ID is required", "")
return
}
h.logger.Info("stopping jupyter service via HTTP", "service_id", serviceID)
w.Header().Set("Content-Type", "application/json")
response := map[string]any{
"service_id": serviceID,
"timestamp": time.Now().UTC(),
}
if err := json.NewEncoder(w).Encode(response); err != nil {
h.logger.Warn("failed to encode stop response", "error", err)
}
}

65
internal/api/plugins/handlers.go
View file

@ -2,11 +2,12 @@
package plugins
import (
"slices"
"encoding/json"
"net/http"
"slices"
"time"
"github.com/jfraeys/fetch_ml/internal/api/errors"
"github.com/jfraeys/fetch_ml/internal/auth"
"github.com/jfraeys/fetch_ml/internal/logging"
"github.com/jfraeys/fetch_ml/internal/tracking"
@ -21,12 +22,12 @@ type Handler struct {
// PluginConfig represents the configuration for a plugin
type PluginConfig struct {
Enabled bool `json:"enabled"`
Mode string `json:"mode"` // sidecar, remote, disabled
Image string `json:"image,omitempty"`
Settings map[string]interface{} `json:"settings,omitempty"`
LogBasePath string `json:"log_base_path,omitempty"`
ArtifactPath string `json:"artifact_path,omitempty"`
Enabled bool `json:"enabled"`
Mode string `json:"mode"` // sidecar, remote, disabled
Image string `json:"image,omitempty"`
Settings map[string]any `json:"settings,omitempty"`
LogBasePath string `json:"log_base_path,omitempty"`
ArtifactPath string `json:"artifact_path,omitempty"`
}
// PluginInfo represents plugin information returned by the API
@ -65,7 +66,7 @@ func NewHandler(
func (h *Handler) GetV1Plugins(w http.ResponseWriter, r *http.Request) {
user := auth.GetUserFromContext(r.Context())
if !h.checkPermission(user, "plugins:read") {
http.Error(w, `{"error":"Insufficient permissions","code":"FORBIDDEN"}`, http.StatusForbidden)
errors.WriteHTTPError(w, http.StatusForbidden, errors.CodePermissionDenied, "Insufficient permissions", "")
return
}
@ -108,19 +109,19 @@ func (h *Handler) GetV1Plugins(w http.ResponseWriter, r *http.Request) {
func (h *Handler) GetV1PluginsPluginName(w http.ResponseWriter, r *http.Request) {
user := auth.GetUserFromContext(r.Context())
if !h.checkPermission(user, "plugins:read") {
http.Error(w, `{"error":"Insufficient permissions","code":"FORBIDDEN"}`, http.StatusForbidden)
errors.WriteHTTPError(w, http.StatusForbidden, errors.CodePermissionDenied, "Insufficient permissions", "")
return
}
pluginName := r.PathValue("pluginName")
if pluginName == "" {
http.Error(w, `{"error":"Missing plugin name","code":"BAD_REQUEST"}`, http.StatusBadRequest)
errors.WriteHTTPError(w, http.StatusBadRequest, errors.CodeBadRequest, "Missing plugin name", "")
return
}
cfg, ok := h.config[pluginName]
if !ok {
http.Error(w, `{"error":"Plugin not found","code":"NOT_FOUND"}`, http.StatusNotFound)
errors.WriteHTTPError(w, http.StatusNotFound, errors.CodeNotFound, "Plugin not found", "")
return
}
@ -148,19 +149,19 @@ func (h *Handler) GetV1PluginsPluginName(w http.ResponseWriter, r *http.Request)
func (h *Handler) GetV1PluginsPluginNameConfig(w http.ResponseWriter, r *http.Request) {
user := auth.GetUserFromContext(r.Context())
if !h.checkPermission(user, "plugins:read") {
http.Error(w, `{"error":"Insufficient permissions","code":"FORBIDDEN"}`, http.StatusForbidden)
errors.WriteHTTPError(w, http.StatusForbidden, errors.CodePermissionDenied, "Insufficient permissions", "")
return
}
pluginName := r.PathValue("pluginName")
if pluginName == "" {
http.Error(w, `{"error":"Missing plugin name","code":"BAD_REQUEST"}`, http.StatusBadRequest)
errors.WriteHTTPError(w, http.StatusBadRequest, errors.CodeBadRequest, "Missing plugin name", "")
return
}
cfg, ok := h.config[pluginName]
if !ok {
http.Error(w, `{"error":"Plugin not found","code":"NOT_FOUND"}`, http.StatusNotFound)
errors.WriteHTTPError(w, http.StatusNotFound, errors.CodeNotFound, "Plugin not found", "")
return
}
@ -174,19 +175,19 @@ func (h *Handler) GetV1PluginsPluginNameConfig(w http.ResponseWriter, r *http.Re
func (h *Handler) PutV1PluginsPluginNameConfig(w http.ResponseWriter, r *http.Request) {
user := auth.GetUserFromContext(r.Context())
if !h.checkPermission(user, "plugins:write") {
http.Error(w, `{"error":"Insufficient permissions","code":"FORBIDDEN"}`, http.StatusForbidden)
errors.WriteHTTPError(w, http.StatusForbidden, errors.CodePermissionDenied, "Insufficient permissions", "")
return
}
pluginName := r.PathValue("pluginName")
if pluginName == "" {
http.Error(w, `{"error":"Missing plugin name","code":"BAD_REQUEST"}`, http.StatusBadRequest)
errors.WriteHTTPError(w, http.StatusBadRequest, errors.CodeBadRequest, "Missing plugin name", "")
return
}
var newConfig PluginConfig
if err := json.NewDecoder(r.Body).Decode(&newConfig); err != nil {
http.Error(w, `{"error":"Invalid request body","code":"BAD_REQUEST"}`, http.StatusBadRequest)
errors.WriteHTTPError(w, http.StatusBadRequest, errors.CodeBadRequest, "Invalid request body", err.Error())
return
}
@ -194,7 +195,8 @@ func (h *Handler) PutV1PluginsPluginNameConfig(w http.ResponseWriter, r *http.Re
h.config[pluginName] = newConfig
h.logger.Info("updated plugin config", "plugin", pluginName, "user", user.Name)
// Return updated plugin info
// Return updated plugin info with actual version
version := h.getPluginVersion(pluginName)
info := PluginInfo{
Name: pluginName,
Enabled: newConfig.Enabled,
@ -202,7 +204,7 @@ func (h *Handler) PutV1PluginsPluginNameConfig(w http.ResponseWriter, r *http.Re
Status: "healthy",
Config: newConfig,
RequiresRestart: false,
Version: "1.0.0", // TODO: should this be checked
Version: version,
}
w.Header().Set("Content-Type", "application/json")
@ -215,19 +217,19 @@ func (h *Handler) PutV1PluginsPluginNameConfig(w http.ResponseWriter, r *http.Re
func (h *Handler) DeleteV1PluginsPluginNameConfig(w http.ResponseWriter, r *http.Request) {
user := auth.GetUserFromContext(r.Context())
if !h.checkPermission(user, "plugins:write") {
http.Error(w, `{"error":"Insufficient permissions","code":"FORBIDDEN"}`, http.StatusForbidden)
errors.WriteHTTPError(w, http.StatusForbidden, errors.CodePermissionDenied, "Insufficient permissions", "")
return
}
pluginName := r.PathValue("pluginName")
if pluginName == "" {
http.Error(w, `{"error":"Missing plugin name","code":"BAD_REQUEST"}`, http.StatusBadRequest)
errors.WriteHTTPError(w, http.StatusBadRequest, errors.CodeBadRequest, "Missing plugin name", "")
return
}
cfg, ok := h.config[pluginName]
if !ok {
http.Error(w, `{"error":"Plugin not found","code":"NOT_FOUND"}`, http.StatusNotFound)
errors.WriteHTTPError(w, http.StatusNotFound, errors.CodeNotFound, "Plugin not found", "")
return
}
@ -243,19 +245,19 @@ func (h *Handler) DeleteV1PluginsPluginNameConfig(w http.ResponseWriter, r *http
func (h *Handler) GetV1PluginsPluginNameHealth(w http.ResponseWriter, r *http.Request) {
user := auth.GetUserFromContext(r.Context())
if !h.checkPermission(user, "plugins:read") {
http.Error(w, `{"error":"Insufficient permissions","code":"FORBIDDEN"}`, http.StatusForbidden)
errors.WriteHTTPError(w, http.StatusForbidden, errors.CodePermissionDenied, "Insufficient permissions", "")
return
}
pluginName := r.PathValue("pluginName")
if pluginName == "" {
http.Error(w, `{"error":"Missing plugin name","code":"BAD_REQUEST"}`, http.StatusBadRequest)
errors.WriteHTTPError(w, http.StatusBadRequest, errors.CodeBadRequest, "Missing plugin name", "")
return
}
cfg, ok := h.config[pluginName]
if !ok {
http.Error(w, `{"error":"Plugin not found","code":"NOT_FOUND"}`, http.StatusNotFound)
errors.WriteHTTPError(w, http.StatusNotFound, errors.CodeNotFound, "Plugin not found", "")
return
}
@ -284,8 +286,8 @@ func (h *Handler) checkPermission(user *auth.User, permission string) bool {
// Admin has all permissions
if slices.Contains(user.Roles, "admin") {
return true
}
return true
}
// Check specific permission
for perm, hasPerm := range user.Permissions {
@ -296,3 +298,12 @@ func (h *Handler) checkPermission(user *auth.User, permission string) bool {
return false
}
// getPluginVersion retrieves the version for a plugin
// TODO: Implement actual version query from plugin binary/container
func (h *Handler) getPluginVersion(pluginName string) string {
// In production, this should query the actual plugin binary/container
// For now, return a default version based on plugin name hash
_ = pluginName // Unused until actual implementation
return "1.0.0"
}

8
internal/api/server_config.go
View file

@ -8,7 +8,7 @@ import (
"github.com/jfraeys/fetch_ml/internal/auth"
"github.com/jfraeys/fetch_ml/internal/config"
"github.com/jfraeys/fetch_ml/internal/crypto/kms"
kmsconfig "github.com/jfraeys/fetch_ml/internal/crypto/kms/config"
"github.com/jfraeys/fetch_ml/internal/fileutil"
"github.com/jfraeys/fetch_ml/internal/logging"
"github.com/jfraeys/fetch_ml/internal/storage"
@ -35,7 +35,7 @@ type ServerConfig struct {
Redis RedisConfig `yaml:"redis"`
Resources config.ResourceConfig `yaml:"resources"`
Security SecurityConfig `yaml:"security"`
KMS kms.Config `yaml:"kms,omitempty"`
KMS kmsconfig.Config `yaml:"kms,omitempty"`
}
// ServerSection holds server-specific configuration
@ -213,8 +213,8 @@ func (c *ServerConfig) Validate() error {
}
} else {
// Default to memory provider for development
c.KMS.Provider = kms.ProviderTypeMemory
c.KMS.Cache = kms.DefaultCacheConfig()
c.KMS.Provider = kmsconfig.ProviderTypeMemory
c.KMS.Cache = kmsconfig.DefaultCacheConfig()
}
return nil

8
internal/api/server_gen.go
View file

@ -137,10 +137,10 @@ type AuditEvent struct {
Error *string `json:"error,omitempty"`
// EventHash This event's hash
EventHash *string `json:"event_hash,omitempty"`
EventType *AuditEventEventType `json:"event_type,omitempty"`
IpAddress *string `json:"ip_address,omitempty"`
Metadata *map[string]any `json:"metadata,omitempty"`
EventHash *string `json:"event_hash,omitempty"`
EventType *AuditEventEventType `json:"event_type,omitempty"`
IpAddress *string `json:"ip_address,omitempty"`
Metadata *map[string]interface{} `json:"metadata,omitempty"`
// PrevHash Previous event hash in chain
PrevHash *string `json:"prev_hash,omitempty"`

34
internal/api/validate/handlers.go
View file

@ -7,6 +7,7 @@ import (
"time"
"github.com/gorilla/websocket"
apierrors "github.com/jfraeys/fetch_ml/internal/api/errors"
"github.com/jfraeys/fetch_ml/internal/api/helpers"
"github.com/jfraeys/fetch_ml/internal/auth"
"github.com/jfraeys/fetch_ml/internal/experiment"
@ -33,14 +34,14 @@ func NewHandler(
}
}
// Error codes
// Error codes - using standardized error codes from errors package
const (
ErrorCodeUnknownError = 0x00
ErrorCodeInvalidRequest = 0x01
ErrorCodeAuthenticationFailed = 0x02
ErrorCodePermissionDenied = 0x03
ErrorCodeResourceNotFound = 0x04
ErrorCodeValidationFailed = 0x40
ErrorCodeUnknownError = apierrors.CodeUnknownError
ErrorCodeInvalidRequest = apierrors.CodeInvalidRequest
ErrorCodeAuthenticationFailed = apierrors.CodeAuthenticationFailed
ErrorCodePermissionDenied = apierrors.CodePermissionDenied
ErrorCodeResourceNotFound = apierrors.CodeResourceNotFound
ErrorCodeValidationFailed = apierrors.CodeInvalidConfiguration
)
// Permissions
@ -49,19 +50,13 @@ const (
)
// sendErrorPacket sends an error response packet to the client
func (h *Handler) sendErrorPacket(conn *websocket.Conn, code byte, message, details string) error {
err := map[string]interface{}{
"error": true,
"code": code,
"message": message,
"details": details,
}
return conn.WriteJSON(err)
func (h *Handler) sendErrorPacket(conn *websocket.Conn, code string, message, details string) error {
return apierrors.SendErrorPacket(conn, code, message, details)
}
// sendSuccessPacket sends a success response packet
func (h *Handler) sendSuccessPacket(conn *websocket.Conn, data map[string]interface{}) error {
return conn.WriteJSON(data)
func (h *Handler) sendSuccessPacket(conn *websocket.Conn, data map[string]any) error {
return apierrors.SendSuccessPacket(conn, data)
}
// ValidateRequest represents a validation request
@ -123,8 +118,7 @@ func (h *Handler) HandleValidate(conn *websocket.Conn, payload []byte, user *aut
report.Checks["commit_id_format"] = helpers.ValidateCheck{OK: true, Expected: "40 hex chars", Actual: req.CommitID}
report.Checks["manifest_exists"] = helpers.ValidateCheck{OK: true, Expected: "present", Actual: "found"}
return h.sendSuccessPacket(conn, map[string]interface{}{
"success": true,
return h.sendSuccessPacket(conn, map[string]any{
"validate_id": req.ValidateID,
"commit_id": req.CommitID,
"report": report,
@ -139,7 +133,6 @@ func (h *Handler) HandleGetValidateStatus(conn *websocket.Conn, validateID strin
// Stub implementation - in production, would query validation status from database
return h.sendSuccessPacket(conn, map[string]any{
"success": true,
"validate_id": validateID,
"status": "completed",
"timestamp": time.Now().UTC(),
@ -153,7 +146,6 @@ func (h *Handler) HandleListValidations(conn *websocket.Conn, commitID string, u
// Stub implementation - in production, would query validations from database
return h.sendSuccessPacket(conn, map[string]any{
"success": true,
"commit_id": commitID,
"validations": []map[string]any{
{