diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..4241a89 --- /dev/null +++ b/.env.example @@ -0,0 +1,14 @@ +POSTGRES_PASSWORD=change-me + +DATABASE_URL=postgres://taskflow:change-me@localhost:5432/taskflow?sslmode=disable +TEST_DATABASE_URL=postgres://taskflow:change-me@localhost:5432/taskflow_test?sslmode=disable + +APP_URL=http://localhost:8081 + +SMTP_ADDRESS=localhost:1025 +SMTP_USERNAME= +SMTP_PASSWORD= +MAIL_FROM=noreply@taskflow.local + +GROQ_API_KEY=replace-with-your-groq-api-key +GROQ_MODEL=openai/gpt-oss-20b diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..902b3e9 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,51 @@ +name: CI + +on: + push: + pull_request: + +permissions: + contents: read + +jobs: + test: + runs-on: ubuntu-latest + + services: + postgres: + image: postgres:18.6-alpine + env: + POSTGRES_USER: taskflow + POSTGRES_PASSWORD: taskflow-test-password + POSTGRES_DB: taskflow_test + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U taskflow -d taskflow_test" + --health-interval 5s + --health-timeout 5s + --health-retries 10 + + env: + TEST_DATABASE_URL: postgres://taskflow:taskflow-test-password@localhost:5432/taskflow_test?sslmode=disable + + steps: + - name: Check out repository + uses: actions/checkout@v4 + + - name: Set up Go + uses: actions/setup-go@v6 + with: + go-version-file: go.mod + + - name: Install migration tool + run: go install -tags 'postgres' github.com/golang-migrate/migrate/v4/cmd/migrate@v4.19.1 + + - name: Apply test migrations + run: $(go env GOPATH)/bin/migrate -path ./migrations -database "$TEST_DATABASE_URL" up + + - name: Run static analysis + run: go vet ./... + + - name: Run tests + run: go test ./... -count=1 diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..40e1595 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 SaadShakeel1 + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md index 3ff135f..8ac77ed 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,139 @@ # TaskFlow -TaskFlow is a collaborative project-management REST API built with Go. +TaskFlow is a collaborative project-management REST API written in Go. It supports authenticated project workspaces, task planning, project members, activity history, and AI-assisted project-risk analysis powered by Groq. -This project is being developed incrementally to learn production-style Go backend development. \ No newline at end of file +## Features + +- Cookie-based authentication with email verification, password hashing, and session logout. +- Projects with owner, editor, and viewer access roles. +- Tasks with assignees, priorities, due dates, estimates, acceptance criteria, state changes, and optimistic locking. +- Task dependencies with duplicate and cycle prevention. +- Filtering, sorting, and paginated project, task, activity, and AI-run lists. +- Immutable activity history for important mutations. +- Background assignment notifications through SMTP/Mailpit. +- Per-client rate limiting, secure JSON decoding, server timeouts, panic recovery, and graceful shutdown. +- Queued AI analysis jobs processed by a Go worker using Groq structured output. + +## Architecture + +```text +HTTP request + -> middleware (security headers, logging, authentication, rate limiting) + -> handler + -> model + -> PostgreSQL + +AI analysis request + -> durable ai_analysis_runs record (queued) + -> background worker claims the run with SKIP LOCKED + -> Groq structured-output request + -> succeeded or failed run persisted for later retrieval +``` + +The application uses the Go standard library's `net/http` router, `database/sql` with PostgreSQL, and SQL migrations. Application code is under `cmd/api` and reusable internal packages live under `internal`. + +## Prerequisites + +- Go 1.26+ +- Docker Desktop and Docker Compose +- PostgreSQL migration CLI: [golang-migrate](https://github.com/golang-migrate/migrate) +- A Groq API key for live AI analysis + +## Local setup + +1. Create your local environment file. + + ```powershell + Copy-Item .env.example .env + ``` + + Set a strong `POSTGRES_PASSWORD`, then use that same value in `DATABASE_URL` and `TEST_DATABASE_URL`. Set `GROQ_API_KEY` locally. Never commit `.env`. + +2. Start PostgreSQL and Mailpit. + + ```powershell + docker compose up -d --wait + ``` + + Mailpit is available at . + +3. Create the test database once. + + ```powershell + docker compose exec db psql -U taskflow -d postgres -c "CREATE DATABASE taskflow_test;" + ``` + +4. Apply migrations to both databases. + + ```powershell + $databaseLine = Get-Content .env | Where-Object { $_ -like 'DATABASE_URL=*' } | Select-Object -First 1 + $env:DATABASE_URL = (($databaseLine -split '=', 2)[1]).Trim() + + $testDatabaseLine = Get-Content .env | Where-Object { $_ -like 'TEST_DATABASE_URL=*' } | Select-Object -First 1 + $env:TEST_DATABASE_URL = (($testDatabaseLine -split '=', 2)[1]).Trim() + + migrate -path ./migrations -database $env:DATABASE_URL up + migrate -path ./migrations -database $env:TEST_DATABASE_URL up + ``` + +5. Run the API. + + ```powershell + go run ./cmd/api + ``` + + The API listens on `http://localhost:8081` by default. Check it with: + + ```powershell + curl.exe http://localhost:8081/health + ``` + +## Tests + +Load the test database URL and run all tests: + +```powershell +$testDatabaseLine = Get-Content .env | Where-Object { $_ -like 'TEST_DATABASE_URL=*' } | Select-Object -First 1 +$env:TEST_DATABASE_URL = (($testDatabaseLine -split '=', 2)[1]).Trim() + +go test ./... -count=1 +go vet ./... +``` + +The real Groq check is deliberately opt-in so normal tests never consume provider usage: + +```powershell +$groqKeyLine = Get-Content .env | Where-Object { $_ -like 'GROQ_API_KEY=*' } | Select-Object -First 1 +$env:GROQ_API_KEY = (($groqKeyLine -split '=', 2)[1]).Trim() +$env:RUN_GROQ_LIVE_TEST = "1" + +go test ./internal/ai -run '^TestGroqAnalyzerLive$' -count=1 +``` + +## API overview + +All routes except registration, login, email verification, health, and greeting require the session cookie created by `POST /sessions`. + +| Area | Endpoints | +| --- | --- | +| Health | `GET /health` | +| Users and sessions | `POST /users`, `POST /users/verify-email`, `POST /sessions`, `DELETE /sessions/current`, `GET /me` | +| Projects | `POST/GET /projects`, `GET/PATCH/DELETE /projects/{id}` | +| Members | `POST/GET /projects/{id}/members`, `PATCH/DELETE /projects/{id}/members/{userID}` | +| Tasks | `POST/GET /projects/{id}/tasks`, `GET/PATCH/DELETE /projects/{id}/tasks/{taskID}` | +| Dependencies | `POST/GET /projects/{id}/tasks/{taskID}/dependencies`, `DELETE /projects/{id}/tasks/{taskID}/dependencies/{dependsOnTaskID}` | +| Activity | `GET /projects/{id}/activity` | +| AI analysis | `POST /projects/{id}/analysis-runs`, `GET /projects/{id}/analysis-runs`, `GET /projects/{id}/analysis-runs/{runID}` | + +`POST /projects/{id}/analysis-runs` returns `202 Accepted`. The background worker processes the saved project snapshot and the client can poll the returned run URL for `queued`, `running`, `succeeded`, or `failed` status. + +## Security notes + +- `.env` is ignored and must never be committed. +- Use a dedicated Groq key with appropriate usage limits. +- The AI worker sends only a scoped project snapshot; it never exposes database credentials, sessions, password hashes, or verification tokens. +- AI findings are advisory. They do not change projects or tasks automatically. + +## Development status + +The backend core is complete. Future work can include deployment, OpenAPI documentation, a frontend, observability, additional worker retry policies, and human-approved AI actions. diff --git a/cmd/api/ai_analysis.go b/cmd/api/ai_analysis.go new file mode 100644 index 0000000..8cdbf6a --- /dev/null +++ b/cmd/api/ai_analysis.go @@ -0,0 +1,210 @@ +package main + +import ( + "encoding/json" + "errors" + "fmt" + "net/http" + "strconv" + "time" + + "github.com/SaadShakeel1/TaskFlow/internal/models" +) + +const ( + defaultAIAnalysisRunPage = 1 + defaultAIAnalysisRunPageSize = 20 + maximumAIAnalysisRunPageSize = 100 +) + +type aiAnalysisRunResponse struct { + ID int64 `json:"id"` + ProjectID int64 `json:"projectId"` + RequestedByUserID int64 `json:"requestedByUserId"` + Status models.AIAnalysisRunStatus `json:"status"` + Result json.RawMessage `json:"result,omitempty"` + Provider *string `json:"provider,omitempty"` + Model *string `json:"model,omitempty"` + AttemptCount int `json:"attemptCount"` + ErrorMessage *string `json:"errorMessage,omitempty"` + StartedAt *time.Time `json:"startedAt,omitempty"` + CompletedAt *time.Time `json:"completedAt,omitempty"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` +} + +type aiAnalysisRunListResponse struct { + Runs []aiAnalysisRunResponse `json:"runs"` + Pagination paginationResponse `json:"pagination"` +} + +func newAIAnalysisRunResponse(run models.AIAnalysisRun) aiAnalysisRunResponse { + return aiAnalysisRunResponse{ + ID: run.ID, + ProjectID: run.ProjectID, + RequestedByUserID: run.RequestedByUserID, + Status: run.Status, + Result: run.Result, + Provider: run.Provider, + Model: run.Model, + AttemptCount: run.AttemptCount, + ErrorMessage: run.ErrorMessage, + StartedAt: run.StartedAt, + CompletedAt: run.CompletedAt, + CreatedAt: run.CreatedAt, + UpdatedAt: run.UpdatedAt, + } +} + +func parseAIAnalysisRunListOptions(r *http.Request) (models.AIAnalysisRunListOptions, error) { + options := models.AIAnalysisRunListOptions{ + Page: defaultAIAnalysisRunPage, + PageSize: defaultAIAnalysisRunPageSize, + } + + query := r.URL.Query() + + if value := query.Get("page"); value != "" { + page, err := strconv.Atoi(value) + if err != nil || page < 1 { + return models.AIAnalysisRunListOptions{}, errors.New("page must be a positive integer") + } + + options.Page = page + } + + if value := query.Get("pageSize"); value != "" { + pageSize, err := strconv.Atoi(value) + if err != nil || pageSize < 1 || pageSize > maximumAIAnalysisRunPageSize { + return models.AIAnalysisRunListOptions{}, errors.New("page size must be between 1 and 100") + } + + options.PageSize = pageSize + } + + return options, nil +} + +func (app *application) createAIAnalysisRunHandler(w http.ResponseWriter, r *http.Request) { + project, user, ok := app.projectForRequest(w, r) + if !ok { + return + } + + canEdit, err := app.userCanEditProject(r.Context(), project, user.ID) + if err != nil { + app.serverError(w, err) + return + } + + if !canEdit { + app.clientError(w, http.StatusForbidden, "project editor permission required") + return + } + + analysisContext, err := app.aiAnalysisRuns.BuildProjectContext(r.Context(), project.ID, user.ID) + if err != nil { + app.serverError(w, err) + return + } + + contextSnapshot, err := json.Marshal(analysisContext) + if err != nil { + app.serverError(w, fmt.Errorf("encode AI analysis context: %w", err)) + return + } + + run, err := app.aiAnalysisRuns.Create(r.Context(), models.AIAnalysisRunCreateInput{ + ProjectID: project.ID, + RequestedByUserID: user.ID, + ContextSnapshot: contextSnapshot, + }) + if err != nil { + app.serverError(w, err) + return + } + + runID := run.ID + app.recordActivity(r.Context(), models.ActivityCreateInput{ + ProjectID: project.ID, + UserID: &user.ID, + Action: "requested", + EntityType: "ai_analysis_run", + EntityID: &runID, + }) + + if err := app.writeJSON(w, http.StatusAccepted, newAIAnalysisRunResponse(run)); err != nil { + app.errorLog.Printf("encode AI analysis run response: %v", err) + } +} + +func (app *application) listAIAnalysisRunsHandler(w http.ResponseWriter, r *http.Request) { + project, user, ok := app.projectForRequest(w, r) + if !ok { + return + } + + options, err := parseAIAnalysisRunListOptions(r) + if err != nil { + app.clientError(w, http.StatusBadRequest, err.Error()) + return + } + + runList, err := app.aiAnalysisRuns.ListByProjectForUser(r.Context(), project.ID, user.ID, options) + if err != nil { + app.serverError(w, err) + return + } + + runs := make([]aiAnalysisRunResponse, 0, len(runList.Runs)) + for _, run := range runList.Runs { + runs = append(runs, newAIAnalysisRunResponse(run)) + } + + response := aiAnalysisRunListResponse{ + Runs: runs, + Pagination: paginationResponse{ + Page: options.Page, + PageSize: options.PageSize, + TotalItems: runList.TotalItems, + TotalPages: (runList.TotalItems + options.PageSize - 1) / options.PageSize, + }, + } + + if err := app.writeJSON(w, http.StatusOK, response); err != nil { + app.errorLog.Printf("encode AI analysis run list response: %v", err) + } +} + +func (app *application) getAIAnalysisRunHandler(w http.ResponseWriter, r *http.Request) { + project, user, ok := app.projectForRequest(w, r) + if !ok { + return + } + + runID, err := strconv.ParseInt(r.PathValue("runID"), 10, 64) + if err != nil || runID < 1 { + app.clientError(w, http.StatusBadRequest, "AI analysis run id must be a positive integer") + return + } + + run, err := app.aiAnalysisRuns.GetByIDForUser(r.Context(), runID, user.ID) + if err != nil { + if errors.Is(err, models.ErrNoRecord) { + app.notFound(w, "AI analysis run not found") + return + } + + app.serverError(w, err) + return + } + + if run.ProjectID != project.ID { + app.notFound(w, "AI analysis run not found") + return + } + + if err := app.writeJSON(w, http.StatusOK, newAIAnalysisRunResponse(run)); err != nil { + app.errorLog.Printf("encode AI analysis run response: %v", err) + } +} diff --git a/cmd/api/ai_analysis_test.go b/cmd/api/ai_analysis_test.go new file mode 100644 index 0000000..22687de --- /dev/null +++ b/cmd/api/ai_analysis_test.go @@ -0,0 +1,72 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "testing" + + "github.com/SaadShakeel1/TaskFlow/internal/testutil" +) + +func TestAIAnalysisRunEndpoints(t *testing.T) { + db := testutil.OpenTestDB(t) + testutil.ResetDatabase(t, db) + t.Cleanup(func() { testutil.ResetDatabase(t, db) }) + + ctx, cancel := context.WithTimeout(context.Background(), databaseTestTimeout) + defer cancel() + + app := newTestApplication(db) + owner, token := createAuthenticatedTestUser(t, app, ctx) + project, err := app.projects.Create(ctx, owner.ID, "AI API project", "Project for AI endpoint tests", nil) + if err != nil { + t.Fatalf("create project: %v", err) + } + + createRequest := httptest.NewRequest(http.MethodPost, fmt.Sprintf("/projects/%d/analysis-runs", project.ID), nil) + createRequest.AddCookie(&http.Cookie{Name: sessionCookieName, Value: token}) + createRecorder := httptest.NewRecorder() + app.routes().ServeHTTP(createRecorder, createRequest) + + if createRecorder.Code != http.StatusAccepted { + t.Fatalf("create status = %d, want %d: %s", createRecorder.Code, http.StatusAccepted, createRecorder.Body.String()) + } + + var created aiAnalysisRunResponse + if err := json.NewDecoder(createRecorder.Body).Decode(&created); err != nil { + t.Fatalf("decode create response: %v", err) + } + + if created.Status != "queued" || created.ProjectID != project.ID { + t.Errorf("created response = %#v", created) + } + + listRequest := httptest.NewRequest(http.MethodGet, fmt.Sprintf("/projects/%d/analysis-runs", project.ID), nil) + listRequest.AddCookie(&http.Cookie{Name: sessionCookieName, Value: token}) + listRecorder := httptest.NewRecorder() + app.routes().ServeHTTP(listRecorder, listRequest) + + if listRecorder.Code != http.StatusOK { + t.Fatalf("list status = %d, want %d", listRecorder.Code, http.StatusOK) + } + + var list aiAnalysisRunListResponse + if err := json.NewDecoder(listRecorder.Body).Decode(&list); err != nil { + t.Fatalf("decode list response: %v", err) + } + if list.Pagination.TotalItems != 1 || len(list.Runs) != 1 || list.Runs[0].ID != created.ID { + t.Errorf("list response = %#v", list) + } + + getRequest := httptest.NewRequest(http.MethodGet, fmt.Sprintf("/projects/%d/analysis-runs/%d", project.ID, created.ID), nil) + getRequest.AddCookie(&http.Cookie{Name: sessionCookieName, Value: token}) + getRecorder := httptest.NewRecorder() + app.routes().ServeHTTP(getRecorder, getRequest) + + if getRecorder.Code != http.StatusOK { + t.Fatalf("get status = %d, want %d", getRecorder.Code, http.StatusOK) + } +} diff --git a/cmd/api/helpers_test.go b/cmd/api/helpers_test.go index e93a6fc..dbebc4f 100644 --- a/cmd/api/helpers_test.go +++ b/cmd/api/helpers_test.go @@ -70,6 +70,7 @@ func newTestApplication(db *sql.DB) *application { tasks: &models.TaskModel{DB: db}, taskDependencies: &models.TaskDependencyModel{DB: db}, activity: &models.ActivityModel{DB: db}, + aiAnalysisRuns: &models.AIAnalysisRunModel{DB: db}, rateLimiter: newRateLimiter(), } } diff --git a/cmd/api/main.go b/cmd/api/main.go index 6ce6326..88dbc5a 100644 --- a/cmd/api/main.go +++ b/cmd/api/main.go @@ -13,6 +13,7 @@ import ( "syscall" "time" + "github.com/SaadShakeel1/TaskFlow/internal/ai" "github.com/SaadShakeel1/TaskFlow/internal/database" "github.com/SaadShakeel1/TaskFlow/internal/mailer" "github.com/SaadShakeel1/TaskFlow/internal/models" @@ -27,11 +28,19 @@ type mailerConfig struct { from string } +type aiConfig struct { + groqAPIKey string + groqModel string + pollInterval time.Duration + processTimeout time.Duration +} + type config struct { addr string appURL string databaseURL string mailer mailerConfig + ai aiConfig sessionTTL time.Duration sessionCookieSecure bool } @@ -41,6 +50,10 @@ type notificationDispatcher interface { Shutdown(ctx context.Context) error } +type aiWorker interface { + Run(context.Context) +} + type application struct { config config db *sql.DB @@ -54,9 +67,11 @@ type application struct { tasks *models.TaskModel taskDependencies *models.TaskDependencyModel activity *models.ActivityModel + aiAnalysisRuns *models.AIAnalysisRunModel mailer mailer.Sender notifications notificationDispatcher rateLimiter *rateLimiter + aiWorker aiWorker } func main() { @@ -76,6 +91,10 @@ func main() { flag.StringVar(&cfg.mailer.smtpUsername, "smtp-username", os.Getenv("SMTP_USERNAME"), "SMTP username") flag.StringVar(&cfg.mailer.smtpPassword, "smtp-password", os.Getenv("SMTP_PASSWORD"), "SMTP password") flag.StringVar(&cfg.mailer.from, "mail-from", os.Getenv("MAIL_FROM"), "Email sender address") + flag.StringVar(&cfg.ai.groqAPIKey, "groq-api-key", os.Getenv("GROQ_API_KEY"), "Groq API key") + flag.StringVar(&cfg.ai.groqModel, "groq-model", os.Getenv("GROQ_MODEL"), "Groq model") + flag.DurationVar(&cfg.ai.pollInterval, "ai-poll-interval", time.Second, "AI worker polling interval") + flag.DurationVar(&cfg.ai.processTimeout, "ai-process-timeout", 60*time.Second, "Maximum duration of one AI analysis") flag.DurationVar(&cfg.sessionTTL, "session-ttl", 7*24*time.Hour, "Session lifetime") flag.BoolVar(&cfg.sessionCookieSecure, "session-cookie-secure", false, "Require HTTPS for session cookie") flag.Parse() @@ -96,6 +115,14 @@ func main() { errorLog.Fatal("MAIL_FROM must be set") } + if cfg.ai.groqAPIKey == "" { + errorLog.Fatal("GROQ_API_KEY must be set") + } + + if cfg.ai.groqModel == "" { + errorLog.Fatal("GROQ_MODEL must be set") + } + db, err := database.Open(cfg.databaseURL) if err != nil { errorLog.Fatal(err) @@ -110,6 +137,27 @@ func main() { }) notificationQueue := notifications.New(emailSender, errorLog) + aiAnalyzer, err := ai.NewGroqAnalyzer(ai.GroqAnalyzerConfig{ + APIKey: cfg.ai.groqAPIKey, + Model: cfg.ai.groqModel, + }) + if err != nil { + errorLog.Fatal(err) + } + + aiAnalysisRuns := &models.AIAnalysisRunModel{DB: db} + aiWorker, err := ai.NewWorker( + aiAnalysisRuns, + aiAnalyzer, + errorLog, + "groq", + cfg.ai.groqModel, + cfg.ai.pollInterval, + cfg.ai.processTimeout, + ) + if err != nil { + errorLog.Fatal(err) + } app := &application{ config: cfg, @@ -124,9 +172,11 @@ func main() { tasks: &models.TaskModel{DB: db}, taskDependencies: &models.TaskDependencyModel{DB: db}, activity: &models.ActivityModel{DB: db}, + aiAnalysisRuns: aiAnalysisRuns, rateLimiter: newRateLimiter(), mailer: emailSender, notifications: notificationQueue, + aiWorker: aiWorker, } server := &http.Server{ diff --git a/cmd/api/routes.go b/cmd/api/routes.go index f8646b3..2cba438 100644 --- a/cmd/api/routes.go +++ b/cmd/api/routes.go @@ -27,6 +27,9 @@ func (app *application) routes() http.Handler { mux.Handle("GET /projects/{id}/tasks/{taskID}", app.requireAuthentication(http.HandlerFunc(app.getTaskHandler))) mux.Handle("GET /projects/{id}/tasks", app.requireAuthentication(http.HandlerFunc(app.listTasksHandler))) mux.Handle("GET /projects/{id}/activity", app.requireAuthentication(http.HandlerFunc(app.listActivityHandler))) + mux.Handle("POST /projects/{id}/analysis-runs", app.requireAuthentication(http.HandlerFunc(app.createAIAnalysisRunHandler))) + mux.Handle("GET /projects/{id}/analysis-runs", app.requireAuthentication(http.HandlerFunc(app.listAIAnalysisRunsHandler))) + mux.Handle("GET /projects/{id}/analysis-runs/{runID}", app.requireAuthentication(http.HandlerFunc(app.getAIAnalysisRunHandler))) mux.Handle("PATCH /projects/{id}/tasks/{taskID}", app.requireAuthentication(http.HandlerFunc(app.updateTaskHandler))) mux.Handle("DELETE /projects/{id}/tasks/{taskID}", app.requireAuthentication(http.HandlerFunc(app.deleteTaskHandler))) mux.Handle("POST /projects/{id}/tasks/{taskID}/dependencies", app.requireAuthentication(http.HandlerFunc(app.createTaskDependencyHandler))) diff --git a/cmd/api/server.go b/cmd/api/server.go index 6e60951..f299227 100644 --- a/cmd/api/server.go +++ b/cmd/api/server.go @@ -13,6 +13,17 @@ const gracefulShutdownTimeout = 10 * time.Second func (app *application) serve(ctx context.Context, server *http.Server, listener net.Listener) error { serverError := make(chan error, 1) + var aiWorkerDone <-chan struct{} + + if app.aiWorker != nil { + workerDone := make(chan struct{}) + aiWorkerDone = workerDone + + go func() { + defer close(workerDone) + app.aiWorker.Run(ctx) + }() + } go func() { serverError <- server.Serve(listener) @@ -50,6 +61,14 @@ func (app *application) serve(ctx context.Context, server *http.Server, listener } } + if aiWorkerDone != nil { + select { + case <-aiWorkerDone: + case <-shutdownCtx.Done(): + return errors.New("shutdown AI worker: timed out") + } + } + app.infoLog.Print("server stopped") return nil diff --git a/internal/ai/groq_analyzer.go b/internal/ai/groq_analyzer.go new file mode 100644 index 0000000..1a96c56 --- /dev/null +++ b/internal/ai/groq_analyzer.go @@ -0,0 +1,202 @@ +package ai + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strings" + "time" + + "github.com/SaadShakeel1/TaskFlow/internal/models" +) + +const ( + groqChatCompletionsURL = "https://api.groq.com/openai/v1/chat/completions" + defaultGroqTimeout = 60 * time.Second + maxGroqErrorBytes = 4 << 10 +) + +type GroqAnalyzerConfig struct { + APIKey string + Model string + Client *http.Client +} + +type GroqAnalyzer struct { + apiKey string + model string + client *http.Client +} + +type groqMessage struct { + Role string `json:"role"` + Content string `json:"content"` +} + +type groqResponseFormat struct { + Type string `json:"type"` + JSONSchema groqJSONSchema `json:"json_schema"` +} + +type groqJSONSchema struct { + Name string `json:"name"` + Strict bool `json:"strict"` + Schema map[string]any `json:"schema"` +} + +type groqChatCompletionRequest struct { + Model string `json:"model"` + Messages []groqMessage `json:"messages"` + Temperature float64 `json:"temperature"` + ResponseFormat groqResponseFormat `json:"response_format"` +} + +type groqChatCompletionResponse struct { + Choices []struct { + Message struct { + Content string `json:"content"` + } `json:"message"` + } `json:"choices"` +} + +func NewGroqAnalyzer(config GroqAnalyzerConfig) (*GroqAnalyzer, error) { + apiKey := strings.TrimSpace(config.APIKey) + if apiKey == "" { + return nil, errors.New("Groq API key is required") + } + + model := strings.TrimSpace(config.Model) + if model == "" { + return nil, errors.New("Groq model is required") + } + + client := config.Client + if client == nil { + client = &http.Client{Timeout: defaultGroqTimeout} + } + + return &GroqAnalyzer{apiKey: apiKey, model: model, client: client}, nil +} + +func (a *GroqAnalyzer) Analyze(ctx context.Context, analysisContext models.ProjectAnalysisContext) (models.AIAnalysisResult, error) { + contextJSON, err := json.Marshal(analysisContext) + if err != nil { + return models.AIAnalysisResult{}, fmt.Errorf("encode project analysis context: %w", err) + } + + requestBody, err := json.Marshal(groqChatCompletionRequest{ + Model: a.model, + Messages: []groqMessage{ + { + Role: "system", + Content: `You are TaskFlow's project-risk analyst. Use only the supplied project snapshot as evidence. + +Identify meaningful delivery risks, blockers, overdue work, dependency bottlenecks, unassigned work, inactivity, or scope risk. + +Do not invent data. Every finding must cite one or more relevant task IDs or activity IDs. Return an empty findings array when the snapshot has no meaningful risk.`, + }, + {Role: "user", Content: string(contextJSON)}, + }, + Temperature: 0.2, + ResponseFormat: groqResponseFormat{ + Type: "json_schema", + JSONSchema: groqJSONSchema{ + Name: "taskflow_project_analysis", + Strict: true, + Schema: projectAnalysisResultSchema(), + }, + }, + }) + if err != nil { + return models.AIAnalysisResult{}, fmt.Errorf("encode Groq request: %w", err) + } + + request, err := http.NewRequestWithContext(ctx, http.MethodPost, groqChatCompletionsURL, bytes.NewReader(requestBody)) + if err != nil { + return models.AIAnalysisResult{}, fmt.Errorf("create Groq request: %w", err) + } + + request.Header.Set("Authorization", "Bearer "+a.apiKey) + request.Header.Set("Content-Type", "application/json") + request.Header.Set("Accept", "application/json") + + response, err := a.client.Do(request) + if err != nil { + return models.AIAnalysisResult{}, fmt.Errorf("call Groq: %w", err) + } + defer response.Body.Close() + + if response.StatusCode != http.StatusOK { + body, readErr := io.ReadAll(io.LimitReader(response.Body, maxGroqErrorBytes)) + if readErr != nil { + return models.AIAnalysisResult{}, fmt.Errorf("Groq returned HTTP %d", response.StatusCode) + } + + message := strings.TrimSpace(string(body)) + if message == "" { + message = http.StatusText(response.StatusCode) + } + + return models.AIAnalysisResult{}, fmt.Errorf("Groq returned HTTP %d: %s", response.StatusCode, message) + } + + var completion groqChatCompletionResponse + if err := json.NewDecoder(response.Body).Decode(&completion); err != nil { + return models.AIAnalysisResult{}, fmt.Errorf("decode Groq response: %w", err) + } + + if len(completion.Choices) != 1 { + return models.AIAnalysisResult{}, errors.New("Groq response must contain exactly one choice") + } + + content := strings.TrimSpace(completion.Choices[0].Message.Content) + if content == "" { + return models.AIAnalysisResult{}, errors.New("Groq response content is empty") + } + + var result models.AIAnalysisResult + if err := json.Unmarshal([]byte(content), &result); err != nil { + return models.AIAnalysisResult{}, fmt.Errorf("decode Groq analysis result: %w", err) + } + + return result, nil +} + +func projectAnalysisResultSchema() map[string]any { + return map[string]any{ + "type": "object", + "properties": map[string]any{ + "summary": map[string]any{"type": "string"}, + "findings": map[string]any{ + "type": "array", + "items": map[string]any{ + "type": "object", + "properties": map[string]any{ + "severity": map[string]any{"type": "string", "enum": []string{"low", "medium", "high", "critical"}}, + "category": map[string]any{"type": "string", "enum": []string{"blocked_task", "overdue_task", "dependency_bottleneck", "unassigned_work", "inactivity", "scope_risk"}}, + "title": map[string]any{"type": "string"}, + "summary": map[string]any{"type": "string"}, + "evidence": map[string]any{ + "type": "object", + "properties": map[string]any{ + "taskIDs": map[string]any{"type": "array", "items": map[string]any{"type": "integer"}}, + "activityIDs": map[string]any{"type": "array", "items": map[string]any{"type": "integer"}}, + }, + "required": []string{"taskIDs", "activityIDs"}, + "additionalProperties": false, + }, + "recommendation": map[string]any{"type": "string"}, + }, + "required": []string{"severity", "category", "title", "summary", "evidence", "recommendation"}, + "additionalProperties": false, + }, + }, + }, + "required": []string{"summary", "findings"}, + "additionalProperties": false, + } +} diff --git a/internal/ai/groq_analyzer_test.go b/internal/ai/groq_analyzer_test.go new file mode 100644 index 0000000..5376cab --- /dev/null +++ b/internal/ai/groq_analyzer_test.go @@ -0,0 +1,127 @@ +package ai + +import ( + "context" + "encoding/json" + "io" + "net/http" + "os" + "strings" + "testing" + + "github.com/SaadShakeel1/TaskFlow/internal/models" +) + +type roundTripperFunc func(*http.Request) (*http.Response, error) + +func (f roundTripperFunc) RoundTrip(request *http.Request) (*http.Response, error) { + return f(request) +} + +func TestGroqAnalyzerAnalyze(t *testing.T) { + resultJSON, err := json.Marshal(models.AIAnalysisResult{ + Summary: "One blocked task needs attention.", + Findings: []models.AIAnalysisFinding{{ + Severity: models.AIInsightSeverityHigh, + Category: models.AIInsightCategoryBlockedTask, + Title: "Blocked task", + Summary: "Task 12 is blocked.", + Evidence: models.AIInsightEvidence{TaskIDs: []int64{12}}, + Recommendation: "Resolve the blocker.", + }}, + }) + if err != nil { + t.Fatal(err) + } + + completionJSON, err := json.Marshal(map[string]any{ + "choices": []any{map[string]any{ + "message": map[string]any{"content": string(resultJSON)}, + }}, + }) + if err != nil { + t.Fatal(err) + } + + analyzer, err := NewGroqAnalyzer(GroqAnalyzerConfig{ + APIKey: "test-key", + Model: "openai/gpt-oss-20b", + Client: &http.Client{Transport: roundTripperFunc(func(request *http.Request) (*http.Response, error) { + if request.Header.Get("Authorization") != "Bearer test-key" { + t.Errorf("authorization = %q", request.Header.Get("Authorization")) + } + + return &http.Response{ + StatusCode: http.StatusOK, + Body: io.NopCloser(strings.NewReader(string(completionJSON))), + Header: make(http.Header), + }, nil + })}, + }) + if err != nil { + t.Fatalf("create analyzer: %v", err) + } + + result, err := analyzer.Analyze(context.Background(), models.ProjectAnalysisContext{SchemaVersion: 1}) + if err != nil { + t.Fatalf("analyze: %v", err) + } + + if result.Summary != "One blocked task needs attention." { + t.Errorf("summary = %q", result.Summary) + } + + if len(result.Findings) != 1 || result.Findings[0].Evidence.TaskIDs[0] != 12 { + t.Errorf("unexpected findings: %#v", result.Findings) + } +} + +func TestGroqAnalyzerLive(t *testing.T) { + if os.Getenv("RUN_GROQ_LIVE_TEST") != "1" { + t.Skip("set RUN_GROQ_LIVE_TEST=1 to call Groq") + } + + apiKey := strings.TrimSpace(os.Getenv("GROQ_API_KEY")) + if apiKey == "" { + t.Fatal("GROQ_API_KEY must be set for the live test") + } + + model := strings.TrimSpace(os.Getenv("GROQ_MODEL")) + if model == "" { + model = "openai/gpt-oss-20b" + } + + analyzer, err := NewGroqAnalyzer(GroqAnalyzerConfig{ + APIKey: apiKey, + Model: model, + }) + if err != nil { + t.Fatalf("create Groq analyzer: %v", err) + } + + result, err := analyzer.Analyze(context.Background(), models.ProjectAnalysisContext{ + SchemaVersion: 1, + Project: models.ProjectAnalysisProject{ + ID: 1, + Name: "Synthetic live-test project", + }, + Tasks: []models.ProjectAnalysisTask{{ + ID: 10, + Title: "Blocked deployment", + Status: models.TaskStatusBlocked, + BlockedReason: pointerTo("Waiting for production credentials"), + Priority: models.TaskPriorityHigh, + }}, + }) + if err != nil { + t.Fatalf("analyze with Groq: %v", err) + } + + if strings.TrimSpace(result.Summary) == "" { + t.Fatal("Groq result summary is empty") + } +} + +func pointerTo(value string) *string { + return &value +} diff --git a/internal/ai/worker.go b/internal/ai/worker.go new file mode 100644 index 0000000..525c311 --- /dev/null +++ b/internal/ai/worker.go @@ -0,0 +1,167 @@ +package ai + +import ( + "context" + "encoding/json" + "errors" + "log" + "strings" + "time" + + "github.com/SaadShakeel1/TaskFlow/internal/models" +) + +type Analyzer interface { + Analyze( + ctx context.Context, + analysisContext models.ProjectAnalysisContext, + ) (models.AIAnalysisResult, error) +} + +type RunModel interface { + ClaimNext( + ctx context.Context, + provider string, + model string, + ) (models.AIAnalysisRun, error) + + MarkSucceeded( + ctx context.Context, + runID int64, + result models.AIAnalysisResult, + ) (models.AIAnalysisRun, error) + + MarkFailed( + ctx context.Context, + runID int64, + failureMessage string, + ) (models.AIAnalysisRun, error) +} + +type Worker struct { + runs RunModel + analyzer Analyzer + logger *log.Logger + provider string + model string + pollInterval time.Duration + processTimeout time.Duration +} + +func NewWorker( + runs RunModel, + analyzer Analyzer, + logger *log.Logger, + provider string, + model string, + pollInterval time.Duration, + processTimeout time.Duration, +) (*Worker, error) { + if runs == nil { + return nil, errors.New("AI analysis run model is required") + } + + if analyzer == nil { + return nil, errors.New("AI analyzer is required") + } + + if logger == nil { + return nil, errors.New("logger is required") + } + + provider = strings.TrimSpace(provider) + if provider == "" { + return nil, errors.New("AI provider is required") + } + + model = strings.TrimSpace(model) + if model == "" { + return nil, errors.New("AI model is required") + } + + if pollInterval <= 0 { + return nil, errors.New("AI worker poll interval must be positive") + } + + if processTimeout <= 0 { + return nil, errors.New("AI worker process timeout must be positive") + } + + return &Worker{ + runs: runs, + analyzer: analyzer, + logger: logger, + provider: provider, + model: model, + pollInterval: pollInterval, + processTimeout: processTimeout, + }, nil +} + +func (w *Worker) Run(ctx context.Context) { + w.processNext(ctx) + + ticker := time.NewTicker(w.pollInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + + case <-ticker.C: + w.processNext(ctx) + } + } +} + +func (w *Worker) processNext(parentContext context.Context) { + ctx, cancel := context.WithTimeout(parentContext, w.processTimeout) + defer cancel() + + run, err := w.runs.ClaimNext(ctx, w.provider, w.model) + if err != nil { + if errors.Is(err, models.ErrNoRecord) { + return + } + + w.logger.Printf("claim AI analysis run: %v", err) + return + } + + var analysisContext models.ProjectAnalysisContext + if err := json.Unmarshal(run.ContextSnapshot, &analysisContext); err != nil { + w.failRun(ctx, run.ID, "stored analysis context is invalid") + return + } + + if analysisContext.SchemaVersion != 1 { + w.failRun(ctx, run.ID, "stored analysis context version is unsupported") + return + } + + result, err := w.analyzer.Analyze(ctx, analysisContext) + if err != nil { + w.failRun(ctx, run.ID, "AI analysis could not be completed") + w.logger.Printf("analyze AI run %d: %v", run.ID, err) + return + } + + if _, err := w.runs.MarkSucceeded(ctx, run.ID, result); err != nil { + w.logger.Printf("mark AI run %d succeeded: %v", run.ID, err) + } +} + +func (w *Worker) failRun( + ctx context.Context, + runID int64, + failureMessage string, +) { + if _, err := w.runs.MarkFailed(ctx, runID, failureMessage); err != nil { + w.logger.Printf("mark AI run %d failed: %v", runID, err) + } +} + +func (w *Worker) ProcessNextForTest(ctx context.Context) { + w.processNext(ctx) +} diff --git a/internal/ai/worker_test.go b/internal/ai/worker_test.go new file mode 100644 index 0000000..e77cea1 --- /dev/null +++ b/internal/ai/worker_test.go @@ -0,0 +1,106 @@ +package ai + +import ( + "context" + "encoding/json" + "errors" + "io" + "log" + "testing" + + "github.com/SaadShakeel1/TaskFlow/internal/models" +) + +type runModelStub struct { + run models.AIAnalysisRun + claimErr error + succeededRunID int64 + failedRunID int64 + failureMessage string +} + +func (s *runModelStub) ClaimNext(context.Context, string, string) (models.AIAnalysisRun, error) { + return s.run, s.claimErr +} + +func (s *runModelStub) MarkSucceeded(_ context.Context, runID int64, _ models.AIAnalysisResult) (models.AIAnalysisRun, error) { + s.succeededRunID = runID + return models.AIAnalysisRun{ID: runID}, nil +} + +func (s *runModelStub) MarkFailed(_ context.Context, runID int64, message string) (models.AIAnalysisRun, error) { + s.failedRunID = runID + s.failureMessage = message + return models.AIAnalysisRun{ID: runID}, nil +} + +type analyzerStub struct { + result models.AIAnalysisResult + err error +} + +func (s analyzerStub) Analyze(context.Context, models.ProjectAnalysisContext) (models.AIAnalysisResult, error) { + return s.result, s.err +} + +func TestWorkerProcessNextMarksRunSucceeded(t *testing.T) { + contextSnapshot, err := json.Marshal(models.ProjectAnalysisContext{SchemaVersion: 1}) + if err != nil { + t.Fatal(err) + } + + runs := &runModelStub{run: models.AIAnalysisRun{ID: 42, ContextSnapshot: contextSnapshot}} + worker, err := NewWorker( + runs, + analyzerStub{result: models.AIAnalysisResult{Summary: "No material delivery risks found."}}, + log.New(io.Discard, "", 0), + "groq", + "test-model", + 1, + 1, + ) + if err != nil { + t.Fatalf("create worker: %v", err) + } + + worker.ProcessNextForTest(context.Background()) + + if runs.succeededRunID != 42 { + t.Errorf("succeeded run ID = %d, want 42", runs.succeededRunID) + } + + if runs.failedRunID != 0 { + t.Errorf("failed run ID = %d, want 0", runs.failedRunID) + } +} + +func TestWorkerProcessNextMarksRunFailedWhenAnalysisFails(t *testing.T) { + contextSnapshot, err := json.Marshal(models.ProjectAnalysisContext{SchemaVersion: 1}) + if err != nil { + t.Fatal(err) + } + + runs := &runModelStub{run: models.AIAnalysisRun{ID: 7, ContextSnapshot: contextSnapshot}} + worker, err := NewWorker( + runs, + analyzerStub{err: errors.New("Groq unavailable")}, + log.New(io.Discard, "", 0), + "groq", + "test-model", + 1, + 1, + ) + if err != nil { + t.Fatalf("create worker: %v", err) + } + + worker.ProcessNextForTest(context.Background()) + + if runs.failedRunID != 7 { + t.Errorf("failed run ID = %d, want 7", runs.failedRunID) + } + + if runs.failureMessage != "AI analysis could not be completed" { + t.Errorf("failure message = %q", runs.failureMessage) + } +} diff --git a/internal/models/ai_analysis_context.go b/internal/models/ai_analysis_context.go new file mode 100644 index 0000000..78fbfe5 --- /dev/null +++ b/internal/models/ai_analysis_context.go @@ -0,0 +1,274 @@ +package models + +import ( + "context" + "encoding/json" + "fmt" + "time" +) + +const ( + aiAnalysisContextSchemaVersion = 1 + aiAnalysisActivityLimit = 100 +) + +type ProjectAnalysisContext struct { + SchemaVersion int `json:"schemaVersion"` + GeneratedAt time.Time `json:"generatedAt"` + Project ProjectAnalysisProject `json:"project"` + Tasks []ProjectAnalysisTask `json:"tasks"` + Dependencies []ProjectAnalysisDependency `json:"dependencies"` + Activities []ProjectAnalysisActivity `json:"activities"` +} + +type ProjectAnalysisProject struct { + ID int64 `json:"id"` + Name string `json:"name"` + Description string `json:"description"` + Status ProjectStatus `json:"status"` + TargetDate *time.Time `json:"targetDate,omitempty"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` +} + +type ProjectAnalysisTask struct { + ID int64 `json:"id"` + AssigneeID *int64 `json:"assigneeID,omitempty"` + Title string `json:"title"` + Description string `json:"description"` + AcceptanceCriteria string `json:"acceptanceCriteria"` + Status TaskStatus `json:"status"` + BlockedReason *string `json:"blockedReason,omitempty"` + Priority TaskPriority `json:"priority"` + DueDate *time.Time `json:"dueDate,omitempty"` + EstimateMinutes *int `json:"estimateMinutes,omitempty"` + StartedAt *time.Time `json:"startedAt,omitempty"` + CompletedAt *time.Time `json:"completedAt,omitempty"` + UpdatedAt time.Time `json:"updatedAt"` +} + +type ProjectAnalysisDependency struct { + TaskID int64 `json:"taskID"` + DependsOnTaskID int64 `json:"dependsOnTaskID"` +} + +type ProjectAnalysisActivity struct { + ID int64 `json:"id"` + UserID *int64 `json:"userID,omitempty"` + Action string `json:"action"` + EntityType string `json:"entityType"` + EntityID *int64 `json:"entityID,omitempty"` + Metadata json.RawMessage `json:"metadata"` + CreatedAt time.Time `json:"createdAt"` +} + +func (m *AIAnalysisRunModel) BuildProjectContext( + ctx context.Context, + projectID int64, + userID int64, +) (ProjectAnalysisContext, error) { + if projectID < 1 || userID < 1 { + return ProjectAnalysisContext{}, fmt.Errorf("project and user IDs must be positive") + } + + project, err := (&ProjectModel{DB: m.DB}).GetByIDForUser(ctx, projectID, userID) + if err != nil { + return ProjectAnalysisContext{}, fmt.Errorf("get project for AI analysis: %w", err) + } + + tasks, err := m.listProjectAnalysisTasks(ctx, projectID) + if err != nil { + return ProjectAnalysisContext{}, err + } + + dependencies, err := m.listProjectAnalysisDependencies(ctx, projectID) + if err != nil { + return ProjectAnalysisContext{}, err + } + + activities, err := m.listProjectAnalysisActivities(ctx, projectID) + if err != nil { + return ProjectAnalysisContext{}, err + } + + return ProjectAnalysisContext{ + SchemaVersion: aiAnalysisContextSchemaVersion, + GeneratedAt: time.Now().UTC(), + Project: ProjectAnalysisProject{ + ID: project.ID, + Name: project.Name, + Description: project.Description, + Status: project.Status, + TargetDate: project.TargetDate, + CreatedAt: project.CreatedAt, + UpdatedAt: project.UpdatedAt, + }, + Tasks: tasks, + Dependencies: dependencies, + Activities: activities, + }, nil +} + +func (m *AIAnalysisRunModel) listProjectAnalysisTasks( + ctx context.Context, + projectID int64, +) ([]ProjectAnalysisTask, error) { + const query = ` + SELECT + id, + assignee_id, + title, + description, + acceptance_criteria, + status, + blocked_reason, + priority, + due_date, + estimate_minutes, + started_at, + completed_at, + updated_at + FROM tasks + WHERE project_id = $1 + ORDER BY id ASC + ` + + rows, err := m.DB.QueryContext(ctx, query, projectID) + if err != nil { + return nil, fmt.Errorf("list tasks for AI analysis: %w", err) + } + defer rows.Close() + + tasks := make([]ProjectAnalysisTask, 0) + + for rows.Next() { + var task ProjectAnalysisTask + + if err := rows.Scan( + &task.ID, + &task.AssigneeID, + &task.Title, + &task.Description, + &task.AcceptanceCriteria, + &task.Status, + &task.BlockedReason, + &task.Priority, + &task.DueDate, + &task.EstimateMinutes, + &task.StartedAt, + &task.CompletedAt, + &task.UpdatedAt, + ); err != nil { + return nil, fmt.Errorf("scan task for AI analysis: %w", err) + } + + tasks = append(tasks, task) + } + + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate tasks for AI analysis: %w", err) + } + + return tasks, nil +} + +func (m *AIAnalysisRunModel) listProjectAnalysisDependencies( + ctx context.Context, + projectID int64, +) ([]ProjectAnalysisDependency, error) { + const query = ` + SELECT + dependency.task_id, + dependency.depends_on_task_id + FROM task_dependencies AS dependency + INNER JOIN tasks AS task ON task.id = dependency.task_id + WHERE task.project_id = $1 + ORDER BY dependency.task_id ASC, dependency.depends_on_task_id ASC + ` + + rows, err := m.DB.QueryContext(ctx, query, projectID) + if err != nil { + return nil, fmt.Errorf("list task dependencies for AI analysis: %w", err) + } + defer rows.Close() + + dependencies := make([]ProjectAnalysisDependency, 0) + + for rows.Next() { + var dependency ProjectAnalysisDependency + + if err := rows.Scan( + &dependency.TaskID, + &dependency.DependsOnTaskID, + ); err != nil { + return nil, fmt.Errorf("scan task dependency for AI analysis: %w", err) + } + + dependencies = append(dependencies, dependency) + } + + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate task dependencies for AI analysis: %w", err) + } + + return dependencies, nil +} + +func (m *AIAnalysisRunModel) listProjectAnalysisActivities( + ctx context.Context, + projectID int64, +) ([]ProjectAnalysisActivity, error) { + const query = ` + SELECT + id, + user_id, + action, + entity_type, + entity_id, + metadata, + created_at + FROM activity_logs + WHERE project_id = $1 + ORDER BY created_at DESC, id DESC + LIMIT $2 + ` + + rows, err := m.DB.QueryContext( + ctx, + query, + projectID, + aiAnalysisActivityLimit, + ) + if err != nil { + return nil, fmt.Errorf("list activities for AI analysis: %w", err) + } + defer rows.Close() + + activities := make([]ProjectAnalysisActivity, 0) + + for rows.Next() { + var activity ProjectAnalysisActivity + var metadata []byte + + if err := rows.Scan( + &activity.ID, + &activity.UserID, + &activity.Action, + &activity.EntityType, + &activity.EntityID, + &metadata, + &activity.CreatedAt, + ); err != nil { + return nil, fmt.Errorf("scan activity for AI analysis: %w", err) + } + + activity.Metadata = json.RawMessage(metadata) + activities = append(activities, activity) + } + + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate activities for AI analysis: %w", err) + } + + return activities, nil +} diff --git a/internal/models/ai_analysis_run.go b/internal/models/ai_analysis_run.go new file mode 100644 index 0000000..ebb8ec4 --- /dev/null +++ b/internal/models/ai_analysis_run.go @@ -0,0 +1,711 @@ +package models + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "strings" + "time" +) + +type AIAnalysisRunStatus string + +const ( + AIAnalysisRunStatusQueued AIAnalysisRunStatus = "queued" + AIAnalysisRunStatusRunning AIAnalysisRunStatus = "running" + AIAnalysisRunStatusSucceeded AIAnalysisRunStatus = "succeeded" + AIAnalysisRunStatusFailed AIAnalysisRunStatus = "failed" + AIAnalysisRunStatusCancelled AIAnalysisRunStatus = "cancelled" +) + +type AIInsightSeverity string + +const ( + AIInsightSeverityLow AIInsightSeverity = "low" + AIInsightSeverityMedium AIInsightSeverity = "medium" + AIInsightSeverityHigh AIInsightSeverity = "high" + AIInsightSeverityCritical AIInsightSeverity = "critical" +) + +type AIInsightCategory string + +const ( + AIInsightCategoryBlockedTask AIInsightCategory = "blocked_task" + AIInsightCategoryOverdueTask AIInsightCategory = "overdue_task" + AIInsightCategoryDependencyBottleneck AIInsightCategory = "dependency_bottleneck" + AIInsightCategoryUnassignedWork AIInsightCategory = "unassigned_work" + AIInsightCategoryInactivity AIInsightCategory = "inactivity" + AIInsightCategoryScopeRisk AIInsightCategory = "scope_risk" +) + +type AIAnalysisResult struct { + Summary string `json:"summary"` + Findings []AIAnalysisFinding `json:"findings"` +} + +type AIAnalysisFinding struct { + Severity AIInsightSeverity `json:"severity"` + Category AIInsightCategory `json:"category"` + Title string `json:"title"` + Summary string `json:"summary"` + Evidence AIInsightEvidence `json:"evidence"` + Recommendation string `json:"recommendation"` +} + +type AIInsightEvidence struct { + TaskIDs []int64 `json:"taskIDs"` + ActivityIDs []int64 `json:"activityIDs"` +} + +type AIAnalysisRun struct { + ID int64 + ProjectID int64 + RequestedByUserID int64 + Status AIAnalysisRunStatus + ContextSchemaVersion int + ContextSnapshot json.RawMessage + Result json.RawMessage + Provider *string + Model *string + AttemptCount int + ErrorMessage *string + StartedAt *time.Time + CompletedAt *time.Time + CreatedAt time.Time + UpdatedAt time.Time +} + +type AIAnalysisRunCreateInput struct { + ProjectID int64 + RequestedByUserID int64 + ContextSnapshot json.RawMessage +} + +type AIAnalysisRunListOptions struct { + Page int + PageSize int +} + +type AIAnalysisRunList struct { + Runs []AIAnalysisRun + TotalItems int +} + +type AIAnalysisRunModel struct { + DB *sql.DB +} + +func (m *AIAnalysisRunModel) Create( + ctx context.Context, + input AIAnalysisRunCreateInput, +) (AIAnalysisRun, error) { + if input.ProjectID < 1 { + return AIAnalysisRun{}, fmt.Errorf("project ID must be positive") + } + + if input.RequestedByUserID < 1 { + return AIAnalysisRun{}, fmt.Errorf("requesting user ID must be positive") + } + + contextSnapshot := input.ContextSnapshot + if len(contextSnapshot) == 0 { + contextSnapshot = json.RawMessage(`{}`) + } + + if !isJSONObject(contextSnapshot) { + return AIAnalysisRun{}, fmt.Errorf("context snapshot must be a JSON object") + } + + const query = ` + INSERT INTO ai_analysis_runs ( + project_id, + requested_by_user_id, + context_snapshot + ) + SELECT $1, $2, $3 + WHERE EXISTS ( + SELECT 1 + FROM projects + WHERE projects.id = $1 + AND ( + projects.user_id = $2 + OR EXISTS ( + SELECT 1 + FROM project_members + WHERE project_members.project_id = projects.id + AND project_members.user_id = $2 + ) + ) + ) + RETURNING + id, + project_id, + requested_by_user_id, + status, + context_schema_version, + context_snapshot, + result, + provider, + model, + attempt_count, + error_message, + started_at, + completed_at, + created_at, + updated_at + ` + + run, err := scanAIAnalysisRun( + m.DB.QueryRowContext( + ctx, + query, + input.ProjectID, + input.RequestedByUserID, + contextSnapshot, + ), + ) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return AIAnalysisRun{}, ErrNoRecord + } + + return AIAnalysisRun{}, fmt.Errorf("create AI analysis run: %w", err) + } + + return run, nil +} + +func (m *AIAnalysisRunModel) GetByIDForUser( + ctx context.Context, + runID int64, + userID int64, +) (AIAnalysisRun, error) { + if runID < 1 { + return AIAnalysisRun{}, fmt.Errorf("AI analysis run ID must be positive") + } + + if userID < 1 { + return AIAnalysisRun{}, fmt.Errorf("user ID must be positive") + } + + const query = ` + SELECT + id, + project_id, + requested_by_user_id, + status, + context_schema_version, + context_snapshot, + result, + provider, + model, + attempt_count, + error_message, + started_at, + completed_at, + created_at, + updated_at + FROM ai_analysis_runs + WHERE id = $1 + AND EXISTS ( + SELECT 1 + FROM projects + WHERE projects.id = ai_analysis_runs.project_id + AND ( + projects.user_id = $2 + OR EXISTS ( + SELECT 1 + FROM project_members + WHERE project_members.project_id = projects.id + AND project_members.user_id = $2 + ) + ) + ) + ` + + run, err := scanAIAnalysisRun( + m.DB.QueryRowContext(ctx, query, runID, userID), + ) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return AIAnalysisRun{}, ErrNoRecord + } + + return AIAnalysisRun{}, fmt.Errorf("get AI analysis run by ID for user: %w", err) + } + + return run, nil +} + +func (m *AIAnalysisRunModel) ClaimNext( + ctx context.Context, + provider string, + model string, +) (AIAnalysisRun, error) { + provider = strings.TrimSpace(provider) + model = strings.TrimSpace(model) + + if provider == "" { + return AIAnalysisRun{}, fmt.Errorf("AI provider is required") + } + + if model == "" { + return AIAnalysisRun{}, fmt.Errorf("AI model is required") + } + + const query = ` + UPDATE ai_analysis_runs + SET + status = 'running', + provider = $1, + model = $2, + attempt_count = attempt_count + 1, + started_at = NOW(), + completed_at = NULL, + error_message = NULL, + updated_at = NOW() + WHERE id = ( + SELECT id + FROM ai_analysis_runs + WHERE status = 'queued' + ORDER BY created_at ASC, id ASC + FOR UPDATE SKIP LOCKED + LIMIT 1 + ) + RETURNING + id, + project_id, + requested_by_user_id, + status, + context_schema_version, + context_snapshot, + result, + provider, + model, + attempt_count, + error_message, + started_at, + completed_at, + created_at, + updated_at + ` + + run, err := scanAIAnalysisRun( + m.DB.QueryRowContext(ctx, query, provider, model), + ) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return AIAnalysisRun{}, ErrNoRecord + } + + return AIAnalysisRun{}, fmt.Errorf("claim next AI analysis run: %w", err) + } + + return run, nil +} + +func (m *AIAnalysisRunModel) MarkSucceeded( + ctx context.Context, + runID int64, + result AIAnalysisResult, +) (AIAnalysisRun, error) { + if runID < 1 { + return AIAnalysisRun{}, fmt.Errorf("AI analysis run ID must be positive") + } + + if err := validateAIAnalysisResult(result); err != nil { + return AIAnalysisRun{}, err + } + + encodedResult, err := json.Marshal(result) + if err != nil { + return AIAnalysisRun{}, fmt.Errorf("encode AI analysis result: %w", err) + } + + const query = ` + UPDATE ai_analysis_runs + SET + status = 'succeeded', + result = $2, + error_message = NULL, + completed_at = NOW(), + updated_at = NOW() + WHERE id = $1 + AND status = 'running' + RETURNING + id, + project_id, + requested_by_user_id, + status, + context_schema_version, + context_snapshot, + result, + provider, + model, + attempt_count, + error_message, + started_at, + completed_at, + created_at, + updated_at + ` + + run, err := scanAIAnalysisRun( + m.DB.QueryRowContext(ctx, query, runID, encodedResult), + ) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return AIAnalysisRun{}, ErrNoRecord + } + + return AIAnalysisRun{}, fmt.Errorf("mark AI analysis run succeeded: %w", err) + } + + return run, nil +} + +func (m *AIAnalysisRunModel) MarkFailed( + ctx context.Context, + runID int64, + failureMessage string, +) (AIAnalysisRun, error) { + if runID < 1 { + return AIAnalysisRun{}, fmt.Errorf("AI analysis run ID must be positive") + } + + failureMessage = strings.TrimSpace(failureMessage) + if failureMessage == "" { + return AIAnalysisRun{}, fmt.Errorf("AI analysis failure message is required") + } + + if len(failureMessage) > 2000 { + return AIAnalysisRun{}, fmt.Errorf("AI analysis failure message must not exceed 2000 characters") + } + + const query = ` + UPDATE ai_analysis_runs + SET + status = 'failed', + result = NULL, + error_message = $2, + completed_at = NOW(), + updated_at = NOW() + WHERE id = $1 + AND status = 'running' + RETURNING + id, + project_id, + requested_by_user_id, + status, + context_schema_version, + context_snapshot, + result, + provider, + model, + attempt_count, + error_message, + started_at, + completed_at, + created_at, + updated_at + ` + + run, err := scanAIAnalysisRun( + m.DB.QueryRowContext(ctx, query, runID, failureMessage), + ) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return AIAnalysisRun{}, ErrNoRecord + } + + return AIAnalysisRun{}, fmt.Errorf("mark AI analysis run failed: %w", err) + } + + return run, nil +} + +func (m *AIAnalysisRunModel) ListByProjectForUser( + ctx context.Context, + projectID int64, + userID int64, + options AIAnalysisRunListOptions, +) (AIAnalysisRunList, error) { + if projectID < 1 || userID < 1 { + return AIAnalysisRunList{}, fmt.Errorf("project and user IDs must be positive") + } + + if options.Page < 1 { + return AIAnalysisRunList{}, fmt.Errorf("page must be at least 1") + } + + if options.PageSize < 1 || options.PageSize > 100 { + return AIAnalysisRunList{}, fmt.Errorf("page size must be between 1 and 100") + } + + const countQuery = ` + SELECT COUNT(*) + FROM ai_analysis_runs AS run + WHERE run.project_id = $1 + AND EXISTS ( + SELECT 1 + FROM projects + WHERE id = run.project_id + AND ( + user_id = $2 + OR EXISTS ( + SELECT 1 + FROM project_members + WHERE project_id = run.project_id + AND user_id = $2 + ) + ) + ) + ` + + var totalItems int + + if err := m.DB.QueryRowContext(ctx, countQuery, projectID, userID).Scan(&totalItems); err != nil { + return AIAnalysisRunList{}, fmt.Errorf("count AI analysis runs by project: %w", err) + } + + offset := (options.Page - 1) * options.PageSize + + const query = ` + SELECT + id, + project_id, + requested_by_user_id, + status, + context_schema_version, + context_snapshot, + result, + provider, + model, + attempt_count, + error_message, + started_at, + completed_at, + created_at, + updated_at + FROM ai_analysis_runs AS run + WHERE run.project_id = $1 + AND EXISTS ( + SELECT 1 + FROM projects + WHERE id = run.project_id + AND ( + user_id = $2 + OR EXISTS ( + SELECT 1 + FROM project_members + WHERE project_id = run.project_id + AND user_id = $2 + ) + ) + ) + ORDER BY run.created_at DESC, run.id DESC + LIMIT $3 OFFSET $4 + ` + + rows, err := m.DB.QueryContext(ctx, query, projectID, userID, options.PageSize, offset) + if err != nil { + return AIAnalysisRunList{}, fmt.Errorf("list AI analysis runs by project: %w", err) + } + defer rows.Close() + + runs := make([]AIAnalysisRun, 0) + + for rows.Next() { + run, err := scanAIAnalysisRun(rows) + if err != nil { + return AIAnalysisRunList{}, fmt.Errorf("scan AI analysis run: %w", err) + } + + runs = append(runs, run) + } + + if err := rows.Err(); err != nil { + return AIAnalysisRunList{}, fmt.Errorf("iterate AI analysis runs: %w", err) + } + + return AIAnalysisRunList{Runs: runs, TotalItems: totalItems}, nil +} + +func validateAIAnalysisResult(result AIAnalysisResult) error { + if summary := strings.TrimSpace(result.Summary); summary == "" { + return fmt.Errorf("AI analysis result summary is required") + } else if len(summary) > 3000 { + return fmt.Errorf("AI analysis result summary must not exceed 3000 characters") + } + + if len(result.Findings) > 50 { + return fmt.Errorf("AI analysis result cannot contain more than 50 findings") + } + + for index, finding := range result.Findings { + if !isValidAIInsightSeverity(finding.Severity) { + return fmt.Errorf("finding %d has invalid severity: %q", index, finding.Severity) + } + + if !isValidAIInsightCategory(finding.Category) { + return fmt.Errorf("finding %d has invalid category: %q", index, finding.Category) + } + + if title := strings.TrimSpace(finding.Title); title == "" { + return fmt.Errorf("finding %d title is required", index) + } else if len(title) > 300 { + return fmt.Errorf("finding %d title must not exceed 300 characters", index) + } + + if summary := strings.TrimSpace(finding.Summary); summary == "" { + return fmt.Errorf("finding %d summary is required", index) + } else if len(summary) > 4000 { + return fmt.Errorf("finding %d summary must not exceed 4000 characters", index) + } + + if recommendation := strings.TrimSpace(finding.Recommendation); recommendation == "" { + return fmt.Errorf("finding %d recommendation is required", index) + } else if len(recommendation) > 2000 { + return fmt.Errorf("finding %d recommendation must not exceed 2000 characters", index) + } + + if len(finding.Evidence.TaskIDs) == 0 && len(finding.Evidence.ActivityIDs) == 0 { + return fmt.Errorf("finding %d must include evidence", index) + } + + if err := validateAIInsightIDs("task", finding.Evidence.TaskIDs); err != nil { + return fmt.Errorf("finding %d: %w", index, err) + } + + if err := validateAIInsightIDs("activity", finding.Evidence.ActivityIDs); err != nil { + return fmt.Errorf("finding %d: %w", index, err) + } + } + + return nil +} + +func isValidAIInsightSeverity(severity AIInsightSeverity) bool { + switch severity { + case AIInsightSeverityLow, + AIInsightSeverityMedium, + AIInsightSeverityHigh, + AIInsightSeverityCritical: + return true + default: + return false + } +} + +func isValidAIInsightCategory(category AIInsightCategory) bool { + switch category { + case AIInsightCategoryBlockedTask, + AIInsightCategoryOverdueTask, + AIInsightCategoryDependencyBottleneck, + AIInsightCategoryUnassignedWork, + AIInsightCategoryInactivity, + AIInsightCategoryScopeRisk: + return true + default: + return false + } +} + +func validateAIInsightIDs(kind string, ids []int64) error { + seen := make(map[int64]struct{}) + + for _, id := range ids { + if id < 1 { + return fmt.Errorf("%s evidence ID must be positive", kind) + } + + if _, exists := seen[id]; exists { + return fmt.Errorf("duplicate %s evidence ID: %d", kind, id) + } + + seen[id] = struct{}{} + } + + return nil +} + +type aiAnalysisRunScanner interface { + Scan(dest ...any) error +} + +func scanAIAnalysisRun(scanner aiAnalysisRunScanner) (AIAnalysisRun, error) { + var run AIAnalysisRun + var contextSnapshot []byte + var result []byte + var provider sql.NullString + var model sql.NullString + var errorMessage sql.NullString + var startedAt sql.NullTime + var completedAt sql.NullTime + + err := scanner.Scan( + &run.ID, + &run.ProjectID, + &run.RequestedByUserID, + &run.Status, + &run.ContextSchemaVersion, + &contextSnapshot, + &result, + &provider, + &model, + &run.AttemptCount, + &errorMessage, + &startedAt, + &completedAt, + &run.CreatedAt, + &run.UpdatedAt, + ) + if err != nil { + return AIAnalysisRun{}, err + } + + run.ContextSnapshot = json.RawMessage(contextSnapshot) + + if result != nil { + run.Result = json.RawMessage(result) + } + + if provider.Valid { + value := provider.String + run.Provider = &value + } + + if model.Valid { + value := model.String + run.Model = &value + } + + if errorMessage.Valid { + value := errorMessage.String + run.ErrorMessage = &value + } + + if startedAt.Valid { + value := startedAt.Time + run.StartedAt = &value + } + + if completedAt.Valid { + value := completedAt.Time + run.CompletedAt = &value + } + + return run, nil +} + +func isJSONObject(value json.RawMessage) bool { + var object map[string]json.RawMessage + + if err := json.Unmarshal(value, &object); err != nil { + return false + } + + return object != nil +} diff --git a/internal/models/ai_analysis_run_test.go b/internal/models/ai_analysis_run_test.go new file mode 100644 index 0000000..798c9d1 --- /dev/null +++ b/internal/models/ai_analysis_run_test.go @@ -0,0 +1,109 @@ +package models_test + +import ( + "context" + "encoding/json" + "errors" + "testing" + + "github.com/SaadShakeel1/TaskFlow/internal/models" + "github.com/SaadShakeel1/TaskFlow/internal/testutil" +) + +func TestAIAnalysisRunModelLifecycle(t *testing.T) { + db := testutil.OpenTestDB(t) + testutil.ResetDatabase(t, db) + t.Cleanup(func() { testutil.ResetDatabase(t, db) }) + + ctx, cancel := context.WithTimeout(context.Background(), databaseTestTimeout) + defer cancel() + + users := &models.UserModel{DB: db} + projects := &models.ProjectModel{DB: db} + runs := &models.AIAnalysisRunModel{DB: db} + + user, err := users.Create(ctx, "ai-run-owner@example.com", "AI Run Owner", "correct-password") + if err != nil { + t.Fatalf("create user: %v", err) + } + + project, err := projects.Create(ctx, user.ID, "AI run project", "Project for AI analysis runs", nil) + if err != nil { + t.Fatalf("create project: %v", err) + } + + snapshot, err := json.Marshal(models.ProjectAnalysisContext{SchemaVersion: 1}) + if err != nil { + t.Fatalf("encode snapshot: %v", err) + } + + created, err := runs.Create(ctx, models.AIAnalysisRunCreateInput{ + ProjectID: project.ID, RequestedByUserID: user.ID, ContextSnapshot: snapshot, + }) + if err != nil { + t.Fatalf("create analysis run: %v", err) + } + + if created.Status != models.AIAnalysisRunStatusQueued { + t.Errorf("status = %q, want queued", created.Status) + } + + claimed, err := runs.ClaimNext(ctx, "groq", "openai/gpt-oss-20b") + if err != nil { + t.Fatalf("claim analysis run: %v", err) + } + + if claimed.ID != created.ID || claimed.Status != models.AIAnalysisRunStatusRunning { + t.Errorf("claimed run = %#v", claimed) + } + + completed, err := runs.MarkSucceeded(ctx, claimed.ID, models.AIAnalysisResult{ + Summary: "No material project risks found.", + }) + if err != nil { + t.Fatalf("mark analysis run succeeded: %v", err) + } + + if completed.Status != models.AIAnalysisRunStatusSucceeded { + t.Errorf("status = %q, want succeeded", completed.Status) + } + + list, err := runs.ListByProjectForUser(ctx, project.ID, user.ID, models.AIAnalysisRunListOptions{Page: 1, PageSize: 20}) + if err != nil { + t.Fatalf("list analysis runs: %v", err) + } + + if list.TotalItems != 1 || len(list.Runs) != 1 || list.Runs[0].ID != created.ID { + t.Errorf("list = %#v", list) + } +} + +func TestAIAnalysisRunModelRejectsOtherProjectMember(t *testing.T) { + db := testutil.OpenTestDB(t) + testutil.ResetDatabase(t, db) + t.Cleanup(func() { testutil.ResetDatabase(t, db) }) + + ctx, cancel := context.WithTimeout(context.Background(), databaseTestTimeout) + defer cancel() + + users := &models.UserModel{DB: db} + projects := &models.ProjectModel{DB: db} + runs := &models.AIAnalysisRunModel{DB: db} + + owner, err := users.Create(ctx, "ai-owner@example.com", "AI Owner", "correct-password") + if err != nil { + t.Fatalf("create owner: %v", err) + } + outsider, err := users.Create(ctx, "ai-outsider@example.com", "AI Outsider", "correct-password") + if err != nil { + t.Fatalf("create outsider: %v", err) + } + project, err := projects.Create(ctx, owner.ID, "Private AI project", "Private", nil) + if err != nil { + t.Fatalf("create project: %v", err) + } + + if _, err := runs.Create(ctx, models.AIAnalysisRunCreateInput{ProjectID: project.ID, RequestedByUserID: outsider.ID, ContextSnapshot: json.RawMessage(`{}`)}); !errors.Is(err, models.ErrNoRecord) { + t.Errorf("create as outsider error = %v, want ErrNoRecord", err) + } +} diff --git a/internal/testutil/postgres.go b/internal/testutil/postgres.go index 1ce2c99..8ae3f94 100644 --- a/internal/testutil/postgres.go +++ b/internal/testutil/postgres.go @@ -83,7 +83,16 @@ func ResetDatabase(t testing.TB, db *sql.DB) { t.Helper() const query = ` - TRUNCATE TABLE activity_logs, task_dependencies, tasks, project_members, projects, sessions, email_verification_tokens, users + TRUNCATE TABLE + ai_analysis_runs, + activity_logs, + task_dependencies, + tasks, + project_members, + projects, + sessions, + email_verification_tokens, + users RESTART IDENTITY ` diff --git a/migrations/000009_create_ai_analysis_runs.down.sql b/migrations/000009_create_ai_analysis_runs.down.sql new file mode 100644 index 0000000..c649636 --- /dev/null +++ b/migrations/000009_create_ai_analysis_runs.down.sql @@ -0,0 +1 @@ +DROP TABLE ai_analysis_runs; diff --git a/migrations/000009_create_ai_analysis_runs.up.sql b/migrations/000009_create_ai_analysis_runs.up.sql new file mode 100644 index 0000000..54fb02d --- /dev/null +++ b/migrations/000009_create_ai_analysis_runs.up.sql @@ -0,0 +1,62 @@ +CREATE TABLE ai_analysis_runs ( + id BIGSERIAL PRIMARY KEY, + + project_id BIGINT NOT NULL + REFERENCES projects(id) ON DELETE CASCADE, + + -- The authenticated user who requested this analysis. + requested_by_user_id BIGINT NOT NULL + REFERENCES users(id) ON DELETE RESTRICT, + + -- The run moves through this lifecycle. + status TEXT NOT NULL DEFAULT 'queued' + CHECK (status IN ('queued', 'running', 'succeeded', 'failed', 'cancelled')), + + -- Versioned, sanitized project evidence sent to the agent. + context_schema_version SMALLINT NOT NULL DEFAULT 1, + context_snapshot JSONB NOT NULL DEFAULT '{}'::jsonb + CHECK (jsonb_typeof(context_snapshot) = 'object'), + + -- Structured result returned by the agent after validation. + result JSONB + CHECK (result IS NULL OR jsonb_typeof(result) = 'object'), + + -- Provider/model names are useful for debugging and cost tracking. + -- They never contain API keys. + provider TEXT, + model TEXT, + + attempt_count INTEGER NOT NULL DEFAULT 0 + CHECK (attempt_count >= 0), + + error_message TEXT, + started_at TIMESTAMPTZ, + completed_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + + -- Active runs have no completion time; terminal runs always do. + CHECK ( + (status IN ('queued', 'running') AND completed_at IS NULL) + OR ( + status IN ('succeeded', 'failed', 'cancelled') + AND completed_at IS NOT NULL + ) + ), + + -- A run cannot finish before it starts. + CHECK ( + completed_at IS NULL + OR started_at IS NULL + OR completed_at >= started_at + ) +); + +-- Shows a project's newest analysis runs efficiently. +CREATE INDEX ai_analysis_runs_project_created_idx + ON ai_analysis_runs(project_id, created_at DESC, id DESC); + +-- Supports a future worker looking for queued work. +CREATE INDEX ai_analysis_runs_pending_idx + ON ai_analysis_runs(created_at ASC, id ASC) + WHERE status IN ('queued', 'running');