package api import ( "context" "encoding/json" "errors" "fmt" "io" "log/slog" "net/http" "net/url" "strconv" "strings" "time" "unicode/utf8" "github.com/bluesky-social/indigo/atproto/identity" "github.com/bluesky-social/indigo/atproto/syntax" "github.com/go-chi/chi/v5" "github.com/google/uuid" "tangled.org/core/api/org_tangled" "tangled.org/core/log" "tangled.org/core/migrator/config" "tangled.org/core/migrator/crypto" "tangled.org/core/migrator/db" "tangled.org/core/migrator/git" migratoroauth "tangled.org/core/migrator/oauth" xrpcerr "tangled.org/core/xrpc/errors" "tangled.org/core/xrpc/serviceauth" ) const ( maxJobsPerTask = 50 maxRequestID = 128 // schema is 140 graphemes, rune count is an upper bound maxDescription = 140 defaultListLimit = 20 maxListLimit = 100 ) type GrantChecker interface { HasSession(context.Context, string) bool } type GrantProber interface { UsableGrant(ctx context.Context, ownerDid string) error } type Server struct { db *db.DB cfg *config.Config signer *serviceauth.Signer serviceAuth *serviceauth.ServiceAuth logger *slog.Logger grants GrantChecker gitRunner git.CommandRunner proxyAddr string } func NewServer( database *db.DB, cfg *config.Config, signer *serviceauth.Signer, serviceAuth *serviceauth.ServiceAuth, grants GrantChecker, logger *slog.Logger, gitRunner git.CommandRunner, proxyAddr string, ) *Server { return &Server{ db: database, cfg: cfg, signer: signer, serviceAuth: serviceAuth, grants: grants, logger: log.SubLogger(logger, "api"), gitRunner: gitRunner, proxyAddr: proxyAddr, } } func (s *Server) gitRunnerOrDefault() git.CommandRunner { if s.gitRunner != nil { return s.gitRunner } return git.RealCommandRunner{} } func (s *Server) Routes() http.Handler { r := chi.NewRouter() r.Get("/.well-known/did.json", s.handleDIDDoc) r.Mount("/xrpc", s.xrpcRoutes()) return r } func (s *Server) xrpcRoutes() http.Handler { r := chi.NewRouter() r.Use(s.cors) r.Use(s.serviceAuth.VerifyServiceAuth) r.Post("/"+org_tangled.TempMigratorCreateTaskNSID, s.handleCreateTask) r.Get("/"+org_tangled.TempMigratorGetTaskNSID, s.handleGetTask) r.Get("/"+org_tangled.TempMigratorDescribeSourceNSID, s.handleDescribeSource) r.Get("/"+org_tangled.TempMigratorListTasksNSID, s.handleListTasks) r.Post("/"+org_tangled.TempMigratorRetryJobNSID, s.handleRetryJob) return r } func (s *Server) cors(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { h := w.Header() h.Set("Access-Control-Allow-Origin", "*") h.Set("Access-Control-Allow-Methods", "GET, POST, OPTIONS") h.Set("Access-Control-Allow-Headers", "Content-Type, Authorization") h.Set("Access-Control-Max-Age", "86400") if r.Method == http.MethodOptions { w.WriteHeader(http.StatusNoContent) return } next.ServeHTTP(w, r) }) } func (s *Server) handleDIDDoc(w http.ResponseWriter, r *http.Request) { if s.signer == nil { s.writeError(w, internalError, http.StatusServiceUnavailable) return } // knot2 resolver requires an atproto_pds entry on did:web issuers doc := s.signer.DIDDocument() doc.Service = []identity.DocService{{ ID: "#atproto_pds", Type: "AtprotoPersonalDataServer", ServiceEndpoint: s.cfg.ServiceURL(), }} w.Header().Set("Content-Type", "application/did+json") _ = json.NewEncoder(w).Encode(doc) } func (s *Server) handleCreateTask(w http.ResponseWriter, r *http.Request) { owner, ok := actorDid(r) if !ok { s.writeError(w, missingActorError, http.StatusUnauthorized) return } var input org_tangled.TempMigratorCreateTask_Input if err := decodeStrictJSON(w, r, &input, 1<<20); err != nil { s.writeError(w, invalidRequest("invalid request body: "+err.Error()), http.StatusBadRequest) return } if input.RequestId == "" { s.writeError(w, invalidRequest("requestId is required"), http.StatusBadRequest) return } if utf8.RuneCountInString(input.RequestId) > maxRequestID { s.writeError(w, invalidRequest(fmt.Sprintf("requestId must be %d characters or fewer", maxRequestID)), http.StatusBadRequest) return } if len(input.Jobs) == 0 { s.writeError(w, invalidRequest("at least one job is required"), http.StatusBadRequest) return } if len(input.Jobs) > maxJobsPerTask { s.writeError(w, invalidRequest(fmt.Sprintf("too many jobs: at most %d allowed, got %d", maxJobsPerTask, len(input.Jobs))), http.StatusBadRequest) return } hasPrivate := false seenNames := make(map[string]struct{}, len(input.Jobs)) for i, job := range input.Jobs { if job == nil { s.writeError(w, invalidRequest(fmt.Sprintf("job %d: missing", i)), http.StatusBadRequest) return } if err := ValidateRepoName(job.Name); err != nil { s.writeError(w, invalidRequest(fmt.Sprintf("job %d: invalid name: %v", i, err)), http.StatusBadRequest) return } name := strings.ToLower(job.Name) if _, twice := seenNames[name]; twice { s.writeError(w, invalidRequest(fmt.Sprintf("job %d: %q is asked for twice", i, job.Name)), http.StatusBadRequest) return } seenNames[name] = struct{}{} if description := DescriptionOf(job); utf8.RuneCountInString(description) > maxDescription { s.writeError(w, invalidRequest(fmt.Sprintf("job %d: description must be %d characters or fewer", i, maxDescription)), http.StatusBadRequest) return } if err := ValidateKnotDID(job.KnotDid); err != nil { s.writeError(w, invalidRequest(fmt.Sprintf("job %d: invalid knotDid: %v", i, err)), http.StatusBadRequest) return } if job.Private != nil && *job.Private { if err := ValidateGitHubSourceURL(job.SourceUrl, s.cfg.GitHubHost); err != nil { s.writeError(w, invalidRequest(fmt.Sprintf("job %d: %v", i, err)), http.StatusBadRequest) return } hasPrivate = true } else if err := ValidateSourceURL(job.SourceUrl); err != nil { s.writeError(w, invalidRequest(fmt.Sprintf("job %d: invalid sourceUrl: %v", i, err)), http.StatusBadRequest) return } } if s.grants == nil || !s.grants.HasSession(r.Context(), owner) { s.writeError(w, grantRequired(s.cfg.ServiceURL(), owner), http.StatusPreconditionRequired) return } if prober, ok := s.grants.(GrantProber); ok { if err := prober.UsableGrant(r.Context(), owner); err != nil { if errors.Is(err, migratoroauth.ErrGrantRequired) { s.writeError(w, grantRequired(s.cfg.ServiceURL(), owner), http.StatusPreconditionRequired) return } // transient probe failures do not block creation; retried at job time s.logger.Error("grant probe failed", "ownerDid", owner, "err", err) } } taskID := uuid.New().String() var encryptedToken *string if hasPrivate { if input.GithubToken == nil || strings.TrimSpace(*input.GithubToken) == "" { s.writeError(w, invalidRequest("githubToken is required when one or more jobs are private"), http.StatusBadRequest) return } aad := crypto.ComputeAAD(owner, taskID, input.RequestId) enc, err := crypto.Encrypt(s.cfg.ParsedMasterKey, *input.GithubToken, aad) if err != nil { s.logger.Error("failed to encrypt githubToken", "err", err) s.writeError(w, internalError, http.StatusInternalServerError) return } encryptedToken = &enc } dbJobs := mapSlice(input.Jobs, func(job *org_tangled.TempMigratorDefs_NewJob) db.CreateJobInput { return db.CreateJobInput{ Name: job.Name, KnotDid: job.KnotDid, SourceURL: job.SourceUrl, Description: DescriptionOf(job), Private: job.Private != nil && *job.Private, } }) batch, jobs, reused, err := s.db.CreateBatch(r.Context(), db.CreateBatchInput{ ID: taskID, OwnerDid: owner, RequestID: input.RequestId, RequestDigest: ComputeRequestDigest(input.Jobs), EncryptedToken: encryptedToken, Jobs: dbJobs, }) if errors.Is(err, db.ErrConflict) { s.writeError(w, taskConflict(), http.StatusConflict) return } if err != nil { s.logger.Error("failed to create task", "err", err, "ownerDid", owner) s.writeError(w, internalError, http.StatusInternalServerError) return } status := http.StatusAccepted if reused { status = http.StatusOK } s.writeTask(w, batch.Task(jobs), status) } func (s *Server) handleGetTask(w http.ResponseWriter, r *http.Request) { owner, ok := actorDid(r) if !ok { s.writeError(w, missingActorError, http.StatusUnauthorized) return } taskID := r.URL.Query().Get("taskId") if taskID == "" { s.writeError(w, invalidRequest("taskId is required"), http.StatusBadRequest) return } batch, jobs, err := s.db.GetBatch(r.Context(), taskID) if errors.Is(err, db.ErrNotFound) || (err == nil && batch.OwnerDid != owner) { s.writeError(w, taskNotFound(), http.StatusNotFound) return } if err != nil { s.logger.Error("failed to read task", "err", err, "taskId", taskID) s.writeError(w, internalError, http.StatusInternalServerError) return } s.writeTask(w, batch.Task(jobs), http.StatusOK) } func (s *Server) handleListTasks(w http.ResponseWriter, r *http.Request) { owner, ok := actorDid(r) if !ok { s.writeError(w, missingActorError, http.StatusUnauthorized) return } limit := defaultListLimit if raw := r.URL.Query().Get("limit"); raw != "" { parsed, err := strconv.Atoi(raw) if err != nil { s.writeError(w, invalidRequest("limit must be an integer"), http.StatusBadRequest) return } limit = min(max(parsed, 1), maxListLimit) } tasks, err := s.db.GetRecentTasksForOwner(r.Context(), owner, limit) if err != nil { s.logger.Error("failed to list tasks", "err", err, "ownerDid", owner) s.writeError(w, internalError, http.StatusInternalServerError) return } out := org_tangled.TempMigratorListTasks_Output{Tasks: mapSlice(tasks, taskView)} s.writeJSON(w, out, http.StatusOK) } func (s *Server) handleRetryJob(w http.ResponseWriter, r *http.Request) { owner, ok := actorDid(r) if !ok { s.writeError(w, missingActorError, http.StatusUnauthorized) return } var input org_tangled.TempMigratorRetryJob_Input if err := decodeStrictJSON(w, r, &input, 64*1024); err != nil { s.writeError(w, invalidRequest("invalid request body: "+err.Error()), http.StatusBadRequest) return } if input.TaskId == "" { s.writeError(w, invalidRequest("taskId is required"), http.StatusBadRequest) return } jobID, err := strconv.ParseInt(input.JobId, 10, 64) if err != nil { s.writeError(w, invalidRequest("jobId must be the id createTask returned for the job"), http.StatusBadRequest) return } job, batch, err := s.db.GetJob(r.Context(), jobID) if errors.Is(err, db.ErrNotFound) { s.writeError(w, taskNotFound(), http.StatusNotFound) return } if err != nil { s.logger.Error("failed to read job", "err", err, "jobId", jobID) s.writeError(w, internalError, http.StatusInternalServerError) return } if batch.ID != input.TaskId || batch.OwnerDid != owner { s.writeError(w, taskNotFound(), http.StatusNotFound) return } if !job.Status.CanRetry() { s.writeError(w, jobNotRetryable(job.Status), http.StatusBadRequest) return } var newEncryptedToken *string if job.Private { expired := batch.CredentialExpiresAt == nil || !batch.CredentialExpiresAt.After(time.Now()) switch { case input.GithubToken != nil && *input.GithubToken != "": aad := crypto.ComputeAAD(batch.OwnerDid, batch.ID, batch.RequestID) enc, err := crypto.Encrypt(s.cfg.ParsedMasterKey, *input.GithubToken, aad) if err != nil { s.logger.Error("failed to encrypt githubToken", "err", err) s.writeError(w, internalError, http.StatusInternalServerError) return } newEncryptedToken = &enc case batch.EncryptedToken == nil || *batch.EncryptedToken == "": s.writeError(w, invalidRequest("githubToken is required to retry a private job"), http.StatusBadRequest) return case expired: s.writeError(w, invalidRequest("the github credential for this task has expired; retry with a fresh one"), http.StatusBadRequest) return } } parentBatch, batchJobs, err := s.db.RequeueJob(r.Context(), jobID, owner, newEncryptedToken) if errors.Is(err, db.ErrNotFound) { s.writeError(w, taskNotFound(), http.StatusNotFound) return } var transition *db.InvalidTransitionError if errors.As(err, &transition) { s.writeError(w, jobNotRetryable(transition.Status), http.StatusBadRequest) return } if err != nil { s.logger.Error("failed to retry job", "err", err, "jobId", jobID) s.writeError(w, internalError, http.StatusInternalServerError) return } s.writeTask(w, parentBatch.Task(batchJobs), http.StatusOK) } func mapSlice[T, U any](items []T, f func(T) U) []U { out := make([]U, len(items)) for i, it := range items { out[i] = f(it) } return out } func taskView(task db.Task) *org_tangled.TempMigratorDefs_Task { updatedAt := task.UpdatedAt.UTC().Format(time.RFC3339) return &org_tangled.TempMigratorDefs_Task{ Id: task.ID, OwnerDid: task.OwnerDid, CreatedAt: task.CreatedAt.UTC().Format(time.RFC3339), UpdatedAt: &updatedAt, Jobs: mapSlice(task.Jobs, jobView), } } func jobView(job db.Job) *org_tangled.TempMigratorDefs_Job { var repoDid *string if job.RepoDid != "" { repoDid = &job.RepoDid } return &org_tangled.TempMigratorDefs_Job{ Id: strconv.FormatInt(job.ID, 10), Name: job.Name, RepoDid: repoDid, KnotDid: job.KnotDid, SourceUrl: job.SourceURL, Private: job.Private, Status: string(job.Status), Attempts: int64(job.Attempts), Error: job.Error, } } func DescriptionOf(job *org_tangled.TempMigratorDefs_NewJob) string { if job == nil || job.Description == nil { return "" } return strings.TrimSpace(*job.Description) } func actorDid(r *http.Request) (string, bool) { did, ok := r.Context().Value(serviceauth.ActorDid).(syntax.DID) if !ok { return "", false } return did.String(), true } func (s *Server) writeTask(w http.ResponseWriter, task db.Task, status int) { s.writeJSON(w, taskView(task), status) } func (s *Server) writeJSON(w http.ResponseWriter, payload any, status int) { w.Header().Set("Content-Type", "application/json") w.Header().Set("Cache-Control", "no-store, no-cache, must-revalidate") w.WriteHeader(status) _ = json.NewEncoder(w).Encode(payload) } func (s *Server) writeError(w http.ResponseWriter, e xrpcerr.XrpcError, status int) { s.writeJSON(w, e, status) } func decodeStrictJSON(w http.ResponseWriter, r *http.Request, dst any, maxBytes int64) error { ct := r.Header.Get("Content-Type") if ct == "" || !strings.HasPrefix(strings.ToLower(ct), "application/json") { return errors.New("Content-Type must be application/json") } if maxBytes <= 0 { maxBytes = 1 << 20 } r.Body = http.MaxBytesReader(w, r.Body, maxBytes) dec := json.NewDecoder(r.Body) dec.DisallowUnknownFields() if err := dec.Decode(dst); err != nil { return err } if dec.More() { return errors.New("request body contains multiple JSON values") } var trailing struct{} if err := dec.Decode(&trailing); err != io.EOF { return errors.New("request body contains unexpected trailing characters") } return nil } func invalidRequest(message string) xrpcerr.XrpcError { return xrpcerr.NewXrpcError(xrpcerr.WithTag("InvalidRequest"), xrpcerr.WithMessage(message)) } func taskNotFound() xrpcerr.XrpcError { return xrpcerr.NewXrpcError( xrpcerr.WithTag("TaskNotFound"), xrpcerr.WithMessage("no such task belongs to you"), ) } func jobNotRetryable(status db.JobStatus) xrpcerr.XrpcError { return xrpcerr.NewXrpcError( xrpcerr.WithTag("JobNotRetryable"), xrpcerr.WithMessage(fmt.Sprintf("cannot retry a job with status %q", status)), ) } func taskConflict() xrpcerr.XrpcError { return xrpcerr.NewXrpcError( xrpcerr.WithTag("TaskConflict"), xrpcerr.WithMessage("requestId was used before with different jobs"), ) } func grantRequired(serviceURL, did string) xrpcerr.XrpcError { start := fmt.Sprintf("%s/oauth/start?did=%s&return_to=%s", strings.TrimSuffix(serviceURL, "/"), did, url.QueryEscape(config.DefaultReturnPath)) return xrpcerr.NewXrpcError( xrpcerr.WithTag("GrantRequired"), xrpcerr.WithMessage(fmt.Sprintf("grant the migrator access at %s", start)), ) } var ( missingActorError = xrpcerr.NewXrpcError( xrpcerr.WithTag("Auth"), xrpcerr.WithMessage("a verified service token is required"), ) internalError = xrpcerr.NewXrpcError( xrpcerr.WithTag("InternalError"), xrpcerr.WithMessage("internal error"), ) )