package methods import ( "context" "encoding/json" "errors" "log/slog" "regexp" "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/config" "github.com/nextlevelbuilder/goclaw/internal/gateway" "github.com/nextlevelbuilder/goclaw/internal/i18n" "github.com/nextlevelbuilder/goclaw/internal/store" "github.com/nextlevelbuilder/goclaw/pkg/protocol" ) var cronSlugRe = regexp.MustCompile(`^[a-z0-9]([a-z0-9-]*[a-z0-9])?$`) // CronMethods handles cron.list, cron.create, cron.update, cron.delete, cron.toggle. type CronMethods struct { service store.CronStore eventBus bus.EventPublisher cfg *config.Config } func NewCronMethods(service store.CronStore, eventBus bus.EventPublisher, cfg *config.Config) *CronMethods { return &CronMethods{service: service, eventBus: eventBus, cfg: cfg} } func (m *CronMethods) Register(router *gateway.MethodRouter) { router.Register(protocol.MethodCronList, m.handleList) router.Register(protocol.MethodCronCreate, m.handleCreate) router.Register(protocol.MethodCronUpdate, m.handleUpdate) router.Register(protocol.MethodCronDelete, m.handleDelete) router.Register(protocol.MethodCronToggle, m.handleToggle) router.Register(protocol.MethodCronStatus, m.handleStatus) router.Register(protocol.MethodCronRun, m.handleRun) router.Register(protocol.MethodCronRuns, m.handleRuns) } func (m *CronMethods) handleList(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) { var params struct { IncludeDisabled bool `json:"includeDisabled"` } if req.Params != nil { json.Unmarshal(req.Params, ¶ms) } userID := "" if !canSeeAll(client.Role(), m.cfg.Gateway.OwnerIDs, client.UserID()) { userID = client.UserID() } jobs := m.service.ListJobs(ctx, params.IncludeDisabled, "", userID) jobs = store.RedactCronJobsCredentialContext(jobs) client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{ "jobs": jobs, "status": m.service.Status(), })) } func (m *CronMethods) handleCreate(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) { locale := store.LocaleFromContext(ctx) var params struct { Name string `json:"name"` Schedule store.CronSchedule `json:"schedule"` Message string `json:"message"` Command *store.CronCommandSpec `json:"command"` // set → deterministic command payload (no LLM) Deliver bool `json:"deliver"` DeliverChannel string `json:"deliverChannel"` DeliverTo string `json:"deliverTo"` WakeHeartbeat bool `json:"wakeHeartbeat"` Stateless *bool `json:"stateless"` // default true for new crons AgentID string `json:"agentId"` } if req.Params != nil { json.Unmarshal(req.Params, ¶ms) } if params.Name == "" { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgRequired, "name"))) return } if !cronSlugRe.MatchString(params.Name) { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidSlug, "name"))) return } isCommand := params.Command != nil if isCommand { if !m.cfg.Cron.CommandEnabled { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrUnauthorized, i18n.T(locale, i18n.MsgCommandCronDisabled))) return } if err := store.ValidateCronCommandSpec(params.Command); err != nil { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, err.Error())) return } } else if params.Message == "" { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgMsgRequired))) return } job, err := m.service.AddJob(ctx, params.Name, params.Schedule, params.Message, params.Deliver, params.DeliverChannel, params.DeliverTo, params.AgentID, client.UserID()) if err != nil { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, err.Error())) return } // Anti-panic guard: a store may return (nil, nil) if the insert succeeded but // the tenant-scoped readback failed (e.g. tenant asymmetry). Never dereference // a nil job below. if job == nil { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, i18n.T(locale, i18n.MsgInternalError, "cron job created but could not be loaded"))) return } // Apply extra fields not in AddJob signature via an immediate patch. // Default stateless=true for new crons (saves tokens); override with explicit false. statelessVal := true if params.Stateless != nil { statelessVal = *params.Stateless } { patch := store.CronJobPatch{Stateless: &statelessVal} if params.WakeHeartbeat { patch.WakeHeartbeat = ¶ms.WakeHeartbeat } if isCommand { patch.Command = params.Command } // Patch under a tenant-consistent context: the job was inserted under // job.TenantID (which falls back to MasterTenantID for nil-tenant / // master-scope connections). Using the raw ctx here would make // lockCronJobForMutation miss the row for nil-tenant callers, silently // dropping the stateless/command patch. updateCtx := store.WithTenantID(ctx, job.TenantID) if updated, pErr := m.service.UpdateJob(updateCtx, job.ID, patch); pErr == nil { job = updated } } client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{ "job": store.RedactCronJobCredentialContext(*job), })) emitAudit(m.eventBus, client, "cron.created", "cron", job.ID) } func (m *CronMethods) handleDelete(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) { locale := store.LocaleFromContext(ctx) var params struct { JobID string `json:"jobId"` } if req.Params != nil { json.Unmarshal(req.Params, ¶ms) } if params.JobID == "" { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgRequired, "jobId"))) return } job, ok := m.service.GetJob(ctx, params.JobID) if !ok { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrNotFound, i18n.T(locale, i18n.MsgJobNotFound))) return } if !canSeeAll(client.Role(), m.cfg.Gateway.OwnerIDs, client.UserID()) { if job.UserID != client.UserID() { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrUnauthorized, i18n.T(locale, i18n.MsgPermissionDenied, "cron job"))) return } } if err := m.service.RemoveJob(ctx, params.JobID); err != nil { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrNotFound, err.Error())) return } client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{ "deleted": true, })) emitAudit(m.eventBus, client, "cron.deleted", "cron", params.JobID) } func (m *CronMethods) handleToggle(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) { locale := store.LocaleFromContext(ctx) var params struct { JobID string `json:"jobId"` Enabled bool `json:"enabled"` } if req.Params != nil { json.Unmarshal(req.Params, ¶ms) } if params.JobID == "" { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgRequired, "jobId"))) return } job, ok := m.service.GetJob(ctx, params.JobID) if !ok { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrNotFound, i18n.T(locale, i18n.MsgJobNotFound))) return } if !canSeeAll(client.Role(), m.cfg.Gateway.OwnerIDs, client.UserID()) { if job.UserID != client.UserID() { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrUnauthorized, i18n.T(locale, i18n.MsgPermissionDenied, "cron job"))) return } } if params.Enabled { if err := store.CheckCronCredentialOwner(ctx, job); err != nil { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrUnauthorized, i18n.T(locale, i18n.MsgPermissionDenied, "cron job credential context"))) return } } if err := m.service.EnableJob(ctx, params.JobID, params.Enabled); err != nil { code := protocol.ErrInvalidRequest if errors.Is(err, store.ErrCronJobNotFound) { code = protocol.ErrNotFound } client.SendResponse(protocol.NewErrorResponse(req.ID, code, err.Error())) return } client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{ "jobId": params.JobID, "enabled": params.Enabled, })) emitAudit(m.eventBus, client, "cron.toggled", "cron", params.JobID) } func (m *CronMethods) handleStatus(_ context.Context, client *gateway.Client, req *protocol.RequestFrame) { client.SendResponse(protocol.NewOKResponse(req.ID, m.service.Status())) } func (m *CronMethods) handleUpdate(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) { locale := store.LocaleFromContext(ctx) var params struct { JobID string `json:"jobId"` ID string `json:"id"` // alias (matching TS) Patch store.CronJobPatch `json:"patch"` } if req.Params != nil { json.Unmarshal(req.Params, ¶ms) } jobID := params.JobID if jobID == "" { jobID = params.ID } if jobID == "" { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgRequired, "jobId"))) return } existing, ok := m.service.GetJob(ctx, jobID) if !ok { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrNotFound, i18n.T(locale, i18n.MsgJobNotFound))) return } if !canSeeAll(client.Role(), m.cfg.Gateway.OwnerIDs, client.UserID()) { if existing.UserID != client.UserID() { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrUnauthorized, i18n.T(locale, i18n.MsgPermissionDenied, "cron job"))) return } } if err := store.CheckCronCredentialOwner(ctx, existing); err != nil { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrUnauthorized, i18n.T(locale, i18n.MsgPermissionDenied, "cron job credential context"))) return } // A command payload on update must clear the same gate as create: command // cron must be enabled and the spec must be valid. Without this, a normal job // could be mutated into a command job (or persisted with an invalid spec) on a // gateway where command cron is disabled. if params.Patch.Command != nil { if !m.cfg.Cron.CommandEnabled { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrUnauthorized, i18n.T(locale, i18n.MsgCommandCronDisabled))) return } if err := store.ValidateCronCommandSpec(params.Patch.Command); err != nil { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, err.Error())) return } } job, err := m.service.UpdateJob(ctx, jobID, params.Patch) if err != nil { code := protocol.ErrInvalidRequest if errors.Is(err, store.ErrCronJobNotFound) { code = protocol.ErrNotFound } client.SendResponse(protocol.NewErrorResponse(req.ID, code, err.Error())) return } client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{ "job": store.RedactCronJobCredentialContext(*job), })) emitAudit(m.eventBus, client, "cron.updated", "cron", jobID) } func (m *CronMethods) handleRun(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) { locale := store.LocaleFromContext(ctx) var params struct { JobID string `json:"jobId"` ID string `json:"id"` Mode string `json:"mode"` // "force" or "due" (default) } if req.Params != nil { json.Unmarshal(req.Params, ¶ms) } jobID := params.JobID if jobID == "" { jobID = params.ID } if jobID == "" { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgRequired, "jobId"))) return } force := params.Mode == "force" // Validate job exists before responding job, ok := m.service.GetJob(ctx, jobID) if !ok { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgJobNotFound))) return } if !canSeeAll(client.Role(), m.cfg.Gateway.OwnerIDs, client.UserID()) { if job.UserID != client.UserID() { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrUnauthorized, i18n.T(locale, i18n.MsgPermissionDenied, "cron job"))) return } } if err := store.CheckCronCredentialOwner(ctx, job); err != nil { client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrUnauthorized, i18n.T(locale, i18n.MsgPermissionDenied, "cron job credential context"))) return } // Respond immediately — job execution happens in background client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{ "ok": true, "ran": true, })) emitAudit(m.eventBus, client, "cron.run", "cron", jobID) // Preserve tenant scope for async execution. tenantID := store.TenantIDFromContext(ctx) go func() { bgCtx := store.WithTenantID(context.Background(), tenantID) if _, _, err := m.service.RunJob(bgCtx, jobID, force); err != nil { slog.Warn("cron.run background error", "jobId", jobID, "error", err) } }() } func (m *CronMethods) handleRuns(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) { var params struct { JobID string `json:"jobId"` ID string `json:"id"` Limit int `json:"limit"` Offset int `json:"offset"` } if req.Params != nil { json.Unmarshal(req.Params, ¶ms) } jobID := params.JobID if jobID == "" { jobID = params.ID } entries, total := m.service.GetRunLog(ctx, jobID, params.Limit, params.Offset) client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{ "entries": entries, "total": total, })) }