package api import ( "context" "database/sql" "encoding/json" "errors" "io" "net/http" "strings" "time" "git.misaka.ren/m1saka/dsh-web/internal/store" ) func (a *API) intents(w http.ResponseWriter, r *http.Request) { u, _, err := a.user(r) if err != nil { fail(w, 401, "authentication required") return } if err = a.s.EnsureIntents(); err != nil { fail(w, 500, "database error") return } switch r.Method { case http.MethodGet: list, err := a.s.Intents(u.ID) if err != nil { fail(w, 500, "database error") return } writeJSON(w, list) case http.MethodPost: var in struct { Name string `json:"name"` Version string `json:"version"` } dec := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4096)) dec.DisallowUnknownFields() if dec.Decode(&in) != nil || dec.Decode(new(any)) != io.EOF || strings.TrimSpace(in.Name) == "" || len(in.Name) > 200 || !validBuildVersion(in.Version) { fail(w, 400, "name and fixed version required") return } if !normalBuildAllowed(in.Version) { fail(w, 400, errBuildUnsupported.Error()) return } i := store.Intent{ID: store.ID(), UID: u.ID, Name: strings.TrimSpace(in.Name), Version: in.Version, Host: a.runtimeHost(r), Status: "waiting"} if err = a.s.ReserveIntent(i); err != nil { if errors.Is(err, store.ErrInstanceLimit) { fail(w, 409, err.Error()) } else { fail(w, 500, "database error") } return } w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusAccepted) writeJSON(w, i) default: fail(w, 405, "method not allowed") } } func (a *API) intent(w http.ResponseWriter, r *http.Request) { u, _, err := a.user(r) if err != nil { fail(w, 401, "authentication required") return } if r.Method != http.MethodDelete { fail(w, 405, "method not allowed") return } if err = a.s.EnsureIntents(); err != nil { fail(w, 500, "database error") return } id := strings.TrimPrefix(r.URL.Path, "/api/intents/") if err = a.s.CancelIntent(id, u.ID); err != nil { if errors.Is(err, store.ErrIntentState) { fail(w, 409, "intent not found or no longer waiting") } else { fail(w, 500, "database error") } return } writeJSON(w, map[string]string{"status": "cancelled"}) } // Call once at startup, after the build queue has recovered interrupted jobs. func (a *API) StartIntents(ctx context.Context) error { if err := a.s.EnsureIntents(); err != nil { return err } if err := a.recoverIntents(ctx); err != nil { return err } go func() { ticker := time.NewTicker(time.Second) defer ticker.Stop() for { select { case <-ctx.Done(): return default: } a.processIntents(ctx) select { case <-ctx.Done(): return case <-ticker.C: } } }() return nil } func (a *API) finishIntent(id, status, detail string) { _, _ = a.s.DB.Exec(`UPDATE creation_intents SET status=?,error=? WHERE id=? AND status IN ('waiting','creating')`, status, detail, id) } func (a *API) recoverIntents(ctx context.Context) error { list, err := a.s.Intents("") if err != nil { return err } for _, in := range list { if in.Status != "creating" { continue } i, err := a.s.GetInstance(in.InstanceID) if errors.Is(err, sql.ErrNoRows) { a.finishIntent(in.ID, "failed", "instance reservation missing after process restart") continue } if err != nil { return err } if err == nil && i.Status == "running" { a.finishIntent(in.ID, "success", "") continue } cleanup, cancel := context.WithTimeout(ctx, 30*time.Second) if err == nil && i.ContainerID == "" { err = a.recoverRuntimeContainer(cleanup, &i) } if err == nil && i.ContainerID != "" { err = a.removeRuntimeContainer(cleanup, i.ContainerID) } if err == nil { err = a.removeRuntimeNetwork(cleanup, in.InstanceID) } cancel() if err != nil { a.s.SetInstance(in.InstanceID, "error") a.finishIntent(in.ID, "failed", "instance creation requires recovery: "+err.Error()) a.s.AddAudit(in.UID, "instance.recovery_error", in.InstanceID, err.Error()) continue } a.s.DeleteInstance(in.InstanceID) _, _ = a.s.DB.Exec("DELETE FROM runtime_creates WHERE instance_id=?", in.InstanceID) _, _ = a.s.DB.Exec("DELETE FROM runtime_networks WHERE instance_id=?", in.InstanceID) a.finishIntent(in.ID, "failed", "instance creation interrupted by process restart") } return nil } func (a *API) processIntents(ctx context.Context) { list, err := a.s.Intents("") if err != nil { return } for _, in := range list { if ctx.Err() != nil { return } if in.Status != "waiting" { continue } if in.BuildID == "" { b, err := a.enqueueBuild(ctx, in.UID, in.Version) if err != nil { a.finishIntent(in.ID, "failed", "build submission failed") continue } if _, err = a.s.DB.Exec(`UPDATE creation_intents SET build_id=? WHERE id=? AND status='waiting'`, b.ID, in.ID); err != nil { continue } in.BuildID = b.ID } var status, imageID, version string err := a.s.DB.QueryRow(`SELECT status,image_id,version FROM image_builds WHERE id=?`, in.BuildID).Scan(&status, &imageID, &version) if err != nil { continue } if version != in.Version || status == "failed" || status == "cancelled" || status == "missing" { a.finishIntent(in.ID, "failed", "build failed; requested version was not created") continue } if status != "success" { continue } work, cancel := context.WithTimeout(ctx, 90*time.Second) verified, err := a.verifiedImagesPoll(work) if err != nil { cancel() continue } if !runtimeImageID.MatchString(imageID) || verified[imageID] != in.Version { cancel() a.finishIntent(in.ID, "failed", "verified image for requested version unavailable") continue } u, err := a.s.UserByID(in.UID) if err != nil { cancel() a.finishIntent(in.ID, "failed", "owner unavailable") continue } i := store.Instance{ID: store.ID(), UID: in.UID, Name: in.Name, Version: in.Version, ImageID: imageID, Status: "creating", Created: time.Now()} if err = a.s.ClaimIntent(in.ID, i); err != nil { cancel() continue } _, err = a.createRuntime(work, u, i, in.Host) cancel() if err != nil { a.finishIntent(in.ID, "failed", err.Error()) } else { a.finishIntent(in.ID, "success", "") } } }