diff --git a/api_docs.md b/api_docs.md index 3867e77..7fcea11 100644 --- a/api_docs.md +++ b/api_docs.md @@ -340,6 +340,7 @@ Note that the `status` field uses integer status codes: Structure: - `data`: Array of cron history records +- `details`: Present on message runs recorded after this field was introduced. Holds the item that was published and where it landed: `url`, `sent`, `failed`, and `manual` (true for runs triggered through [`/api/message/retry`](#apimessageretry)). Absent for older records and for collect runs. - `pagination`: Pagination metadata object containing: - `total_count`: Total number of records matching the filters - `current_page`: Current page number @@ -368,7 +369,12 @@ Structure: "name": "message", "timestamp": "2024-03-15T10:10:00Z", "status": 2, - "output": "Message sent to: telegram. Failed: bluesky" + "output": "Message sent to: telegram. Failed: bluesky", + "details": { + "url": "https://github.com/resemble-ai/chatterbox", + "sent": ["telegram"], + "failed": ["bluesky"] + } } ], "pagination": { @@ -469,6 +475,70 @@ curl -H "Authorization: Bearer " \ - 401: Unauthorized - Invalid or missing Bearer token - 500: Internal Server Error - Database or server error +### /api/message/retry + +**Endpoint:** `/api/message/retry` + +**Method:** `POST` + +**Description:** Re-send an already published repository to the integrations that did not receive it. + +A message run marks the repository as posted as soon as **any** integration succeeds, which drops the item out of the publication queue. The connectors that failed can therefore never recover it on the next run — this endpoint is the way to finish such a partial publication by hand. + +The repository text is fetched per integration in that integration's configured `text_language`, and one image is generated for the whole retry when any integration has `socialify_image` enabled. No Pushover notification is sent: a manual retry is already being watched by whoever triggered it. + +Retries are serialised — a second one waits for the first, so a double-clicked button cannot publish twice. An item that is still unposted is marked as posted only when **every** requested integration succeeded; marking it after a partial retry would drop it out of the queue again, which is the failure this endpoint repairs. + +**Curl Example:** + +```bash +curl -X POST \ + 'http://localhost:8080/api/message/retry' \ + -H 'Authorization: Bearer ' \ + -H 'Content-Type: application/json' \ + -d '{ + "apis": ["threads"], + "url": "https://github.com/resemble-ai/chatterbox" + }' +``` + +**Request Parameters:** + +| Parameter | Type | Required | Description | +| --------- | -------- | -------- | --------------------------------------------------------------------------------------------------------------- | +| `apis` | string[] | Yes | Names of the integrations to send to, as configured in`/api/api-configs`. Blanks and duplicates are ignored. | +| `url` | string | No | Repository to publish. When omitted the most recently published repository is used, which is only a guess at what a partial run consumed: a run that failed for *every* integration never marked its item as posted, so the guess resolves to the previous one. Callers that know the item — the dashboard reads it from the run details — should always pass it. | + +**Response Structure:** + +- `url`: The repository that was published +- `status`: `0` (nothing sent), `1` (all sent), `2` (partially sent) — the same codes as cron history +- `message`: The text recorded in cron history +- `succeeded` / `failed`: Integration names per outcome +- `outcomes`: Per-integration detail, with an `error` string for every failure + +**Response Example:** + +```json +{ + "url": "https://github.com/resemble-ai/chatterbox", + "status": 1, + "message": "Manual retry: https://github.com/resemble-ai/chatterbox sent to: threads", + "succeeded": ["threads"], + "failed": null, + "outcomes": [{ "api_name": "threads", "success": true }] +} +``` + +Every retry is recorded in cron history under the `message` name with `details.manual = true`, so [`/api/cron-history`](#apicron-history) shows it alongside scheduled runs. + +**Status Codes:** + +- 200: The retry ran. Individual integration failures are reported in `outcomes`, not in the status code +- 400: Bad Request - Invalid body or an empty `apis` list +- 401: Unauthorized - Invalid or missing Bearer token +- 500: Internal Server Error - API configurations not loaded, or the repository could not be resolved + ### /api/api-configs **Endpoint:** `/api/api-configs` diff --git a/cmd/main.go b/cmd/main.go index 001867b..f0aadb4 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -111,6 +111,7 @@ func main() { mux.Handle("/api/collect-settings", middleware.LoggingMiddleware(middleware.CorsMiddleware(middleware.AuthMiddleware(http.HandlerFunc(cronAPI.HandleCollectSettings))))) mux.Handle("/api/prompt-settings", middleware.LoggingMiddleware(middleware.CorsMiddleware(middleware.AuthMiddleware(http.HandlerFunc(cronAPI.HandlePromptSettings))))) mux.Handle("/api/cron-history", middleware.LoggingMiddleware(middleware.CorsMiddleware(middleware.AuthMiddleware(http.HandlerFunc(cronAPI.GetCronHistory))))) + mux.Handle("/api/message/retry", middleware.LoggingMiddleware(middleware.CorsMiddleware(middleware.AuthMiddleware(http.HandlerFunc(cronAPI.RetryMessagePost))))) mux.Handle("/api/api-configs", middleware.LoggingMiddleware(middleware.CorsMiddleware(middleware.AuthMiddleware(http.HandlerFunc(cronAPI.HandleAPIConfigs))))) mux.Handle("/api/api-configs/", middleware.LoggingMiddleware(middleware.CorsMiddleware(middleware.AuthMiddleware(http.HandlerFunc(cronAPI.HandleAPIConfig))))) diff --git a/internal/models/cron.go b/internal/models/cron.go index b2a3930..0817df5 100644 --- a/internal/models/cron.go +++ b/internal/models/cron.go @@ -21,3 +21,10 @@ type UpdateScheduleRequest struct { type UpdateStatusRequest struct { IsActive bool `json:"is_active"` } + +// RetryMessageRequest asks for a repository to be re-sent to the named APIs. +// URL is optional: when empty the most recently published repository is used. +type RetryMessageRequest struct { + APIs []string `json:"apis"` + URL string `json:"url"` +} diff --git a/internal/models/cron_history.go b/internal/models/cron_history.go index 5ab9975..3f4e53b 100644 --- a/internal/models/cron_history.go +++ b/internal/models/cron_history.go @@ -3,10 +3,21 @@ package models import "time" type CronHistory struct { - Name string `json:"name"` - Timestamp time.Time `json:"timestamp"` - Success int `json:"status"` - Output string `json:"output,omitempty"` + Name string `json:"name"` + Timestamp time.Time `json:"timestamp"` + Success int `json:"status"` + Output string `json:"output,omitempty"` + Details *MessageRunDetails `json:"details,omitempty"` +} + +// MessageRunDetails records which item a message run published and where it +// landed, so a partial run can be re-sent to the connectors that missed it +// without re-parsing the human-readable output. +type MessageRunDetails struct { + URL string `json:"url,omitempty"` + Sent []string `json:"sent,omitempty"` + Failed []string `json:"failed,omitempty"` + Manual bool `json:"manual,omitempty"` } type PaginationMetadata struct { diff --git a/internal/repository/endpoints.go b/internal/repository/endpoints.go new file mode 100644 index 0000000..b76f801 --- /dev/null +++ b/internal/repository/endpoints.go @@ -0,0 +1,76 @@ +package repository + +import ( + "bytes" + "context" + "io" + "net/http" + "os" + "strconv" + "strings" + "time" +) + +// client carries no timeout of its own: it is applied per request from the +// environment, so nothing about the content-alchemist connection depends on when +// this package was imported. +var client = &http.Client{} + +func getContentAlchemistTimeout() time.Duration { + timeoutStr := os.Getenv("CONTENT_ALCHEMIST_TIMEOUT") + if timeoutStr == "" { + return 30 * time.Second + } + + timeoutSeconds, err := strconv.Atoi(timeoutStr) + if err != nil { + return 30 * time.Second + } + + return time.Duration(timeoutSeconds) * time.Second +} + +// doRequest sends a request to content-alchemist under the configured timeout. +func doRequest(req *http.Request) (*http.Response, error) { + ctx, cancel := context.WithTimeout(req.Context(), getContentAlchemistTimeout()) + defer cancel() + + resp, err := client.Do(req.WithContext(ctx)) + if err != nil { + return nil, err + } + + // The deferred cancel must not close the body before the caller reads it, so + // the body is buffered and the connection released here. + body, readErr := io.ReadAll(resp.Body) + resp.Body.Close() + if readErr != nil { + return nil, readErr + } + resp.Body = io.NopCloser(bytes.NewReader(body)) + + return resp, nil +} + +// The content-alchemist location is resolved per call rather than at import +// time, so it does not depend on when the environment was populated and can be +// pointed at a stub in tests. +func contentAlchemistURL(path string) string { + return strings.TrimRight(os.Getenv("CONTENT_ALCHEMIST_URL"), "/") + path +} + +func getRepositoryURL() string { + return contentAlchemistURL("/think-root/api/get-repository/") +} + +func updatePostedURL() string { + return contentAlchemistURL("/think-root/api/update-posted/") +} + +func deleteRepositoryURL() string { + return contentAlchemistURL("/think-root/api/delete-repository/") +} + +func authorizationHeader() string { + return "Bearer " + os.Getenv("CONTENT_ALCHEMIST_BEARER") +} diff --git a/internal/repository/get_repository.go b/internal/repository/get_repository.go index 776fb69..fc8c9d2 100644 --- a/internal/repository/get_repository.go +++ b/internal/repository/get_repository.go @@ -1,19 +1,16 @@ package repository import ( + "bytes" "encoding/json" "fmt" + "io" "net/http" - "os" "strings" ) -func init() { - getRepositoryUrl = os.Getenv("CONTENT_ALCHEMIST_URL") + "/think-root/api/get-repository/" - bearerToken = "Bearer " + os.Getenv("CONTENT_ALCHEMIST_BEARER") -} - -type repo struct { +// Item is a single repository as returned by content-alchemist. +type Item struct { ID int `json:"id"` Posted bool `json:"posted"` URL string `json:"url"` @@ -27,7 +24,7 @@ type repositoryResponse struct { All int `json:"all"` Posted int `json:"posted"` Unposted int `json:"unposted"` - Items []repo `json:"items"` + Items []Item `json:"items"` Page int `json:"page"` PageSize int `json:"page_size"` TotalPages int `json:"total_pages"` @@ -37,36 +34,84 @@ type repositoryResponse struct { Status string `json:"status"` } +type getRepositoryRequest struct { + Limit int `json:"limit,omitempty"` + Posted *bool `json:"posted,omitempty"` + SortOrder string `json:"sort_order,omitempty"` + SortBy string `json:"sort_by,omitempty"` + TextLanguage string `json:"text_language,omitempty"` + URL string `json:"url,omitempty"` +} + func GetRepository(limit int, posted bool, sort_order, sort_by string, textLanguage ...string) (*repositoryResponse, error) { var lang string if len(textLanguage) > 0 && textLanguage[0] != "" { lang = textLanguage[0] } - response, err := makeRepositoryRequest(limit, posted, sort_order, sort_by, lang) + return makeRepositoryRequest(getRepositoryRequest{ + Limit: limit, + Posted: &posted, + SortOrder: sort_order, + SortBy: sort_by, + TextLanguage: lang, + }) +} + +// GetRepositoryByURL fetches one repository by its url regardless of its posted +// state. Used when a specific publication has to be re-sent to a connector, so +// the item must not be looked up through the publication queue. +func GetRepositoryByURL(url, textLanguage string) (*Item, error) { + if strings.TrimSpace(url) == "" { + return nil, fmt.Errorf("repository url is required") + } + + response, err := makeRepositoryRequest(getRepositoryRequest{ + URL: url, + TextLanguage: textLanguage, + }) if err != nil { return nil, err } - if response.Status == "error" && strings.Contains(response.Message, "no text available for language") && lang != "uk" { - return makeRepositoryRequest(limit, posted, sort_order, sort_by, "uk") + if len(response.Data.Items) == 0 { + return nil, fmt.Errorf("repository %s not found", url) + } + + item := response.Data.Items[0] + + // An older content-alchemist silently ignores the url filter and answers with + // the head of the queue instead. Posting that item would publish the wrong + // repository without any error, so refuse anything we did not ask for. + if item.URL != url { + return nil, fmt.Errorf("content-alchemist returned repository %s instead of %s", item.URL, url) } - return response, nil + return &item, nil } -func makeRepositoryRequest(limit int, posted bool, sort_order, sort_by, textLanguage string) (*repositoryResponse, error) { - payloadStr := fmt.Sprintf(`{ - "limit": %d, - "posted": %t, - "sort_order": "%s", - "sort_by": "%s", - "text_language": "%s" - }`, limit, posted, sort_order, sort_by, textLanguage) +// GetLatestPostedRepository returns the most recently published repository. +func GetLatestPostedRepository(textLanguage string) (*Item, error) { + response, err := GetRepository(1, true, "DESC", "date_posted", textLanguage) + if err != nil { + return nil, err + } - payload := strings.NewReader(payloadStr) + if len(response.Data.Items) == 0 { + return nil, fmt.Errorf("no published repositories found") + } + + item := response.Data.Items[0] + return &item, nil +} - req, err := http.NewRequest(http.MethodPost, getRepositoryUrl, payload) +func makeRepositoryRequest(payload getRepositoryRequest) (*repositoryResponse, error) { + body, err := json.Marshal(payload) + if err != nil { + return nil, fmt.Errorf("error encoding request: %w", err) + } + + req, err := http.NewRequest(http.MethodPost, getRepositoryURL(), bytes.NewReader(body)) if err != nil { return nil, fmt.Errorf("error creating request: %w", err) } @@ -75,19 +120,57 @@ func makeRepositoryRequest(limit int, posted bool, sort_order, sort_by, textLang "Accept": {"*/*"}, "Connection": {"keep-alive"}, "Content-Type": {"application/json"}, - "Authorization": {bearerToken}, + "Authorization": {authorizationHeader()}, } - resp, err := client.Do(req) + resp, err := doRequest(req) if err != nil { return nil, fmt.Errorf("error sending request: %w", err) } defer resp.Body.Close() + respBody, err := io.ReadAll(resp.Body) + if err != nil { + return nil, fmt.Errorf("error reading response: %w", err) + } + + // A rejected request must surface as an error. Decoding it as a normal + // payload leaves Items empty, which callers used to report as "no items + // available" and hid the real cause. + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { + return nil, fmt.Errorf("content-alchemist error (status %d): %s", resp.StatusCode, errorDetail(respBody)) + } + var response repositoryResponse - if err := json.NewDecoder(resp.Body).Decode(&response); err != nil { + if err := json.Unmarshal(respBody, &response); err != nil { return nil, fmt.Errorf("error decoding response: %w", err) } + if response.Status == "error" { + return nil, fmt.Errorf("content-alchemist error: %s", errorDetail(respBody)) + } + return &response, nil } + +// errorDetail pulls the message out of a content-alchemist error envelope and +// falls back to the raw body, which is plain text for auth and rate-limit +// rejections. +func errorDetail(body []byte) string { + var envelope struct { + Message string `json:"message"` + } + if err := json.Unmarshal(body, &envelope); err == nil && envelope.Message != "" { + return envelope.Message + } + + detail := strings.TrimSpace(string(body)) + if detail == "" { + return "empty response body" + } + if len(detail) > 300 { + return detail[:300] + "…" + } + + return detail +} diff --git a/internal/repository/get_repository_test.go b/internal/repository/get_repository_test.go index da8bee1..a0ece03 100644 --- a/internal/repository/get_repository_test.go +++ b/internal/repository/get_repository_test.go @@ -1,23 +1,16 @@ package repository import ( + "encoding/json" "net/http" "net/http/httptest" "testing" ) -func TestGetRepository(t *testing.T) { - originalClient := client - originalURL := getRepositoryUrl - originalBearer := bearerToken - - bearerToken = "Bearer test-token" +const testBearerToken = "Bearer test-token" - defer func() { - client = originalClient - getRepositoryUrl = originalURL - bearerToken = originalBearer - }() +func TestGetRepository(t *testing.T) { + t.Setenv("CONTENT_ALCHEMIST_BEARER", "test-token") tests := []struct { name string @@ -74,7 +67,36 @@ func TestGetRepository(t *testing.T) { posted: true, serverResponse: `{"error": "Internal Server Error"}`, statusCode: http.StatusInternalServerError, - wantErr: false, + wantErr: true, + expectedAll: 0, + }, + { + // A rejected request must not be mistaken for an empty queue. + name: "rejected request", + limit: 1, + sort_by: "publication_queue", + sort_order: "ASC", + serverResponse: `{"message":"Invalid language code: xx","status":"error"}`, + statusCode: http.StatusBadRequest, + wantErr: true, + expectedAll: 0, + }, + { + // 401 and 429 come back as plain text from the middleware. + name: "plain text rejection", + limit: 1, + serverResponse: "Too Many Requests", + statusCode: http.StatusTooManyRequests, + wantErr: true, + expectedAll: 0, + }, + { + // 200 with an error envelope must not look like a successful fetch. + name: "error envelope with 200", + limit: 1, + serverResponse: `{"message":"Failed to fetch repositories","status":"error"}`, + statusCode: http.StatusOK, + wantErr: true, expectedAll: 0, }, } @@ -90,8 +112,8 @@ func TestGetRepository(t *testing.T) { t.Errorf("Expected Content-Type header to be application/json") } - if r.Header.Get("Authorization") != bearerToken { - t.Errorf("Expected Authorization header to be %s, got %s", bearerToken, r.Header.Get("Authorization")) + if r.Header.Get("Authorization") != testBearerToken { + t.Errorf("Expected Authorization header to be %s, got %s", testBearerToken, r.Header.Get("Authorization")) } w.WriteHeader(tt.statusCode) @@ -99,8 +121,7 @@ func TestGetRepository(t *testing.T) { })) defer server.Close() - getRepositoryUrl = server.URL - client = server.Client() + t.Setenv("CONTENT_ALCHEMIST_URL", server.URL) resp, err := GetRepository(tt.limit, tt.posted, tt.sort_order, tt.sort_by) if (err != nil) != tt.wantErr { @@ -114,3 +135,80 @@ func TestGetRepository(t *testing.T) { }) } } + +func TestGetRepositoryByURL(t *testing.T) { + t.Setenv("CONTENT_ALCHEMIST_BEARER", "test-token") + + const wantURL = "https://github.com/resemble-ai/chatterbox" + + tests := []struct { + name string + requestURL string + serverResponse string + statusCode int + wantErr bool + }{ + { + name: "returns the requested repository", + requestURL: wantURL, + serverResponse: `{"data":{"items":[{"id":1327,"posted":true,` + + `"url":"https://github.com/resemble-ai/chatterbox","text":"English text"}]},"status":"ok"}`, + statusCode: http.StatusOK, + wantErr: false, + }, + { + // An alchemist that does not know the url filter answers with the head + // of the queue; publishing that would post the wrong repository. + name: "rejects a different repository", + requestURL: wantURL, + serverResponse: `{"data":{"items":[{"id":1,"posted":false,` + + `"url":"https://github.com/open-webui/open-webui","text":"Other text"}]},"status":"ok"}`, + statusCode: http.StatusOK, + wantErr: true, + }, + { + name: "reports an empty result", + requestURL: wantURL, + serverResponse: `{"data":{"items":[]},"status":"ok"}`, + statusCode: http.StatusOK, + wantErr: true, + }, + { + name: "requires a url", + requestURL: " ", + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var body getRepositoryRequest + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + t.Errorf("failed to decode request body: %v", err) + } + if body.URL != tt.requestURL { + t.Errorf("expected url %q in request, got %q", tt.requestURL, body.URL) + } + if body.TextLanguage != "en" { + t.Errorf("expected text_language en, got %q", body.TextLanguage) + } + + w.WriteHeader(tt.statusCode) + w.Write([]byte(tt.serverResponse)) + })) + defer server.Close() + + t.Setenv("CONTENT_ALCHEMIST_URL", server.URL) + + item, err := GetRepositoryByURL(tt.requestURL, "en") + if (err != nil) != tt.wantErr { + t.Fatalf("GetRepositoryByURL() error = %v, wantErr %v", err, tt.wantErr) + } + + if err == nil && item.URL != wantURL { + t.Errorf("GetRepositoryByURL() returned %q, want %q", item.URL, wantURL) + } + }) + } +} diff --git a/internal/repository/update_repository.go b/internal/repository/update_repository.go index f269ea4..cb5b14f 100644 --- a/internal/repository/update_repository.go +++ b/internal/repository/update_repository.go @@ -1,14 +1,10 @@ package repository import ( + "bytes" "encoding/json" "fmt" "net/http" - "os" - "strconv" - "strings" - "sync" - "time" ) type updateResponse struct { @@ -16,43 +12,18 @@ type updateResponse struct { Status string `json:"status"` } -var ( - client *http.Client - updatePostedUrl string - getRepositoryUrl string - bearerToken string - once sync.Once -) - -func getContentAlchemistTimeout() time.Duration { - timeoutStr := os.Getenv("CONTENT_ALCHEMIST_TIMEOUT") - if timeoutStr == "" { - return 30 * time.Second - } - - timeoutSeconds, err := strconv.Atoi(timeoutStr) - if err != nil { - return 30 * time.Second - } - - return time.Duration(timeoutSeconds) * time.Second -} - -func init() { - once.Do(func() { - updatePostedUrl = os.Getenv("CONTENT_ALCHEMIST_URL") + "/think-root/api/update-posted/" - getRepositoryUrl = os.Getenv("CONTENT_ALCHEMIST_URL") + "/think-root/api/get-repository/" - bearerToken = "Bearer " + os.Getenv("CONTENT_ALCHEMIST_BEARER") - client = &http.Client{ - Timeout: getContentAlchemistTimeout(), - } - }) +type updatePostedRequest struct { + URL string `json:"url"` + Posted bool `json:"posted"` } func UpdateRepositoryPosted(url string, posted bool) (bool, error) { - payload := strings.NewReader(fmt.Sprintf(`{"url":"%s","posted":%t}`, url, posted)) + payload, err := json.Marshal(updatePostedRequest{URL: url, Posted: posted}) + if err != nil { + return false, fmt.Errorf("error encoding request: %w", err) + } - req, err := http.NewRequest(http.MethodPatch, updatePostedUrl, payload) + req, err := http.NewRequest(http.MethodPatch, updatePostedURL(), bytes.NewReader(payload)) if err != nil { return false, fmt.Errorf("error creating request: %w", err) } @@ -61,10 +32,10 @@ func UpdateRepositoryPosted(url string, posted bool) (bool, error) { "Accept": {"*/*"}, "Connection": {"keep-alive"}, "Content-Type": {"application/json"}, - "Authorization": {bearerToken}, + "Authorization": {authorizationHeader()}, } - resp, err := client.Do(req) + resp, err := doRequest(req) if err != nil { return false, fmt.Errorf("error making request: %w", err) } diff --git a/internal/repository/update_repository_test.go b/internal/repository/update_repository_test.go index 1ba7cb0..94855f6 100644 --- a/internal/repository/update_repository_test.go +++ b/internal/repository/update_repository_test.go @@ -5,22 +5,11 @@ import ( "io" "net/http" "net/http/httptest" - "os" "testing" ) func TestUpdateRepositoryPosted(t *testing.T) { - originalClient := client - originalURL := updatePostedUrl - originalBearer := bearerToken - - bearerToken = "Bearer test-token" - - defer func() { - client = originalClient - updatePostedUrl = originalURL - bearerToken = originalBearer - }() + t.Setenv("CONTENT_ALCHEMIST_BEARER", "test-token") tests := []struct { name string @@ -85,8 +74,8 @@ func TestUpdateRepositoryPosted(t *testing.T) { t.Errorf("Expected Content-Type header to be application/json") } - if r.Header.Get("Authorization") != bearerToken { - t.Errorf("Expected Authorization header to be %s, got %s", bearerToken, r.Header.Get("Authorization")) + if r.Header.Get("Authorization") != testBearerToken { + t.Errorf("Expected Authorization header to be %s, got %s", testBearerToken, r.Header.Get("Authorization")) } if tt.checkRequest { @@ -103,8 +92,7 @@ func TestUpdateRepositoryPosted(t *testing.T) { })) defer server.Close() - updatePostedUrl = server.URL - client = server.Client() + t.Setenv("CONTENT_ALCHEMIST_URL", server.URL) result, err := UpdateRepositoryPosted(tt.url, tt.posted) if (err != nil) != tt.wantErr { @@ -119,48 +107,30 @@ func TestUpdateRepositoryPosted(t *testing.T) { } } -func TestEnvironmentVariables(t *testing.T) { - originalURL := os.Getenv("CONTENT_ALCHEMIST_URL") - originalBearer := os.Getenv("CONTENT_ALCHEMIST_BEARER") - - defer func() { - os.Setenv("CONTENT_ALCHEMIST_URL", originalURL) - os.Setenv("CONTENT_ALCHEMIST_BEARER", originalBearer) - }() - - originalUpdateURL := updatePostedUrl - originalGetURL := getRepositoryUrl - originalToken := bearerToken - - updatePostedUrl = "" - getRepositoryUrl = "" - bearerToken = "" - - os.Setenv("CONTENT_ALCHEMIST_URL", "https://test.example.com") - os.Setenv("CONTENT_ALCHEMIST_BEARER", "test-token") +func TestEndpointsFollowEnvironment(t *testing.T) { + t.Setenv("CONTENT_ALCHEMIST_URL", "https://test.example.com") + t.Setenv("CONTENT_ALCHEMIST_BEARER", "test-token") - var _ = &http.Client{} - - updatePostedUrl = os.Getenv("CONTENT_ALCHEMIST_URL") + "/think-root/api/update-posted/" - getRepositoryUrl = os.Getenv("CONTENT_ALCHEMIST_URL") + "/think-root/api/get-repository/" - bearerToken = "Bearer " + os.Getenv("CONTENT_ALCHEMIST_BEARER") - - expectedUpdateURL := "https://test.example.com/think-root/api/update-posted/" - if updatePostedUrl != expectedUpdateURL { - t.Errorf("Expected updatePostedUrl to be %s, got %s", expectedUpdateURL, updatePostedUrl) + tests := []struct { + name string + got string + want string + }{ + {name: "update posted", got: updatePostedURL(), want: "https://test.example.com/think-root/api/update-posted/"}, + {name: "get repository", got: getRepositoryURL(), want: "https://test.example.com/think-root/api/get-repository/"}, + {name: "delete repository", got: deleteRepositoryURL(), want: "https://test.example.com/think-root/api/delete-repository/"}, + {name: "authorization", got: authorizationHeader(), want: "Bearer test-token"}, } - expectedGetURL := "https://test.example.com/think-root/api/get-repository/" - if getRepositoryUrl != expectedGetURL { - t.Errorf("Expected getRepositoryUrl to be %s, got %s", expectedGetURL, getRepositoryUrl) + for _, tt := range tests { + if tt.got != tt.want { + t.Errorf("%s = %s, want %s", tt.name, tt.got, tt.want) + } } - expectedToken := "Bearer test-token" - if bearerToken != expectedToken { - t.Errorf("Expected bearerToken to be %s, got %s", expectedToken, bearerToken) + // A trailing slash in the configured base must not double up in the path. + t.Setenv("CONTENT_ALCHEMIST_URL", "https://test.example.com/") + if got, want := getRepositoryURL(), "https://test.example.com/think-root/api/get-repository/"; got != want { + t.Errorf("get repository with trailing slash = %s, want %s", got, want) } - - updatePostedUrl = originalUpdateURL - getRepositoryUrl = originalGetURL - bearerToken = originalToken } diff --git a/internal/repository/validate_repository.go b/internal/repository/validate_repository.go index 43cfe6b..ec7aeaa 100644 --- a/internal/repository/validate_repository.go +++ b/internal/repository/validate_repository.go @@ -1,20 +1,12 @@ package repository import ( + "bytes" "encoding/json" "fmt" "net/http" - "os" - "strings" ) -var ( - deleteRepositoryUrl string -) - -func init() { - deleteRepositoryUrl = os.Getenv("CONTENT_ALCHEMIST_URL") + "/think-root/api/delete-repository/" -} func ValidateRepositoryURL(url string) (int, error) { req, err := http.NewRequest(http.MethodHead, url, nil) if err != nil { @@ -22,7 +14,7 @@ func ValidateRepositoryURL(url string) (int, error) { } httpClient := &http.Client{ - Timeout: client.Timeout, + Timeout: getContentAlchemistTimeout(), CheckRedirect: func(req *http.Request, via []*http.Request) error { return http.ErrUseLastResponse }, @@ -43,9 +35,12 @@ type deleteResponse struct { } func DeleteRepository(url string) (bool, error) { - payload := strings.NewReader(fmt.Sprintf(`{"url":"%s"}`, url)) + payload, err := json.Marshal(map[string]string{"url": url}) + if err != nil { + return false, fmt.Errorf("error encoding request: %w", err) + } - req, err := http.NewRequest(http.MethodDelete, deleteRepositoryUrl, payload) + req, err := http.NewRequest(http.MethodDelete, deleteRepositoryURL(), bytes.NewReader(payload)) if err != nil { return false, fmt.Errorf("error creating request: %w", err) } @@ -54,10 +49,10 @@ func DeleteRepository(url string) (bool, error) { "Accept": {"*/*"}, "Connection": {"keep-alive"}, "Content-Type": {"application/json"}, - "Authorization": {bearerToken}, + "Authorization": {authorizationHeader()}, } - resp, err := client.Do(req) + resp, err := doRequest(req) if err != nil { return false, fmt.Errorf("error making request: %w", err) } diff --git a/internal/repository/validate_repository_test.go b/internal/repository/validate_repository_test.go index 097c58a..7e95f75 100644 --- a/internal/repository/validate_repository_test.go +++ b/internal/repository/validate_repository_test.go @@ -70,17 +70,7 @@ func TestValidateRepositoryURL_InvalidURL(t *testing.T) { } func TestDeleteRepository(t *testing.T) { - originalClient := client - originalURL := deleteRepositoryUrl - originalBearer := bearerToken - - bearerToken = "Bearer test-token" - - defer func() { - client = originalClient - deleteRepositoryUrl = originalURL - bearerToken = originalBearer - }() + t.Setenv("CONTENT_ALCHEMIST_BEARER", "test-token") tests := []struct { name string @@ -140,8 +130,8 @@ func TestDeleteRepository(t *testing.T) { t.Errorf("Expected Content-Type header to be application/json") } - if r.Header.Get("Authorization") != bearerToken { - t.Errorf("Expected Authorization header to be %s, got %s", bearerToken, r.Header.Get("Authorization")) + if r.Header.Get("Authorization") != testBearerToken { + t.Errorf("Expected Authorization header to be %s, got %s", testBearerToken, r.Header.Get("Authorization")) } w.WriteHeader(tt.statusCode) @@ -149,8 +139,7 @@ func TestDeleteRepository(t *testing.T) { })) defer server.Close() - deleteRepositoryUrl = server.URL - client = server.Client() + t.Setenv("CONTENT_ALCHEMIST_URL", server.URL) result, err := DeleteRepository(tt.url) if (err != nil) != tt.wantErr { diff --git a/internal/schedule/message-publish.go b/internal/schedule/message-publish.go new file mode 100644 index 0000000..cfb14c0 --- /dev/null +++ b/internal/schedule/message-publish.go @@ -0,0 +1,306 @@ +package schedule + +import ( + "content-maestro/internal/api" + "content-maestro/internal/models" + "content-maestro/internal/repository" + "content-maestro/internal/socialify" + "content-maestro/internal/store" + "content-maestro/internal/utils" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + "sync" + "time" +) + +// ErrInvalidRetryRequest marks a retry that was rejected because of its input, +// so callers can answer with 400 instead of 500. +var ErrInvalidRetryRequest = errors.New("invalid retry request") + +// retrySocialifyConfig keeps image generation short: a manual retry answers an +// HTTP request, and the cron default (5 attempts, 20s apart) would block it for +// well over a minute. +var retrySocialifyConfig = socialify.RetryConfig{ + MaxRetries: 2, + RetryInterval: 3 * time.Second, +} + +const ( + imageDir = "./tmp/gh_project_img" + retryImageDir = imageDir + "/retry" +) + +// retryMutex serialises manual retries. Two of them publishing the same item +// concurrently - a double-clicked button, or two open dashboards - would post +// twice to the same connector. +var retryMutex sync.Mutex + +// imageURLPath turns a local image path into the path it is served under +// /images/, so an image kept in a subdirectory stays reachable. +func imageURLPath(imageName string) string { + relative, err := filepath.Rel(filepath.Clean(imageDir), filepath.Clean(imageName)) + if err != nil || strings.HasPrefix(relative, "..") { + return filepath.Base(imageName) + } + + return filepath.ToSlash(relative) +} + +// publishItem sends one repository to one configured API. Shared by the message +// cron and by manual retries so both build requests the same way. +func publishItem(apiName string, endpoint api.APIEndpoint, item repository.Item, imageName string) (*api.APIResponse, error) { + var req api.RequestConfig + + commonFields := map[string]string{ + "text": item.Text, + "url": item.URL, + } + + switch strings.ToLower(endpoint.ContentType) { + case "multipart": + req = api.RequestConfig{ + APIName: apiName, + FormFields: commonFields, + } + + if endpoint.SocialifyImage && imageName != "" { + req.FileFields = map[string]string{ + "image": imageName, + } + } + case "json": + req = api.RequestConfig{ + APIName: apiName, + JSONBody: map[string]any{"text": item.Text, "url": item.URL}, + } + + if endpoint.SocialifyImage && imageName != "" { + publicURL := os.Getenv("PUBLIC_URL") + if publicURL != "" { + req.JSONBody["image_url"] = fmt.Sprintf("%s/images/%s", publicURL, imageURLPath(imageName)) + } else { + log.Error("PUBLIC_URL not set, cannot generate image_url for API %s", apiName) + } + } + default: + req = api.RequestConfig{ + APIName: apiName, + JSONBody: map[string]any{"text": item.Text, "url": item.URL}, + } + } + + return api.ExecuteRequest(req) +} + +// RetryOutcome is the per-connector result of a manual retry. +type RetryOutcome struct { + APIName string `json:"api_name"` + Success bool `json:"success"` + Error string `json:"error,omitempty"` +} + +// RetryResult is the response of a manual retry. +type RetryResult struct { + URL string `json:"url"` + Status int `json:"status"` + Message string `json:"message"` + Succeeded []string `json:"succeeded"` + Failed []string `json:"failed"` + Outcomes []RetryOutcome `json:"outcomes"` +} + +// RetryMessagePost re-sends one repository to the named APIs. It exists because a +// partially successful message run marks the item as posted, which drops it out +// of the publication queue even though some connectors never received it. +// +// When url is empty the most recently published repository is used. That is only +// a guess at what a partial run consumed, so callers that know the item - the +// dashboard reads it from the run details - should always pass it explicitly. +func RetryMessagePost(st store.StoreInterface, apiNames []string, url string) (*RetryResult, error) { + apiConfigs := api.GetAPIConfigs() + if apiConfigs == nil { + return nil, fmt.Errorf("API configurations not loaded") + } + + requested, err := normalizeAPINames(apiNames) + if err != nil { + return nil, err + } + + retryMutex.Lock() + defer retryMutex.Unlock() + + // Pin the target before contacting any connector so every API in this call + // publishes the same repository. + url = strings.TrimSpace(url) + itemPosted := false + if url == "" { + latest, err := repository.GetLatestPostedRepository("") + if err != nil { + return nil, fmt.Errorf("failed to resolve the latest published repository: %w", err) + } + url = latest.URL + itemPosted = latest.Posted + } + + result := &RetryResult{URL: url} + + // One image per retry, not one per connector: the cron shares a single image + // across all of them, and each generation is a separate upstream fetch. + imageName := "" + defer func() { + if imageName == "" { + return + } + if err := os.Remove(imageName); err != nil && !os.IsNotExist(err) { + log.Errorf("Failed to remove retry image %s: %v", imageName, err) + } + }() + + for _, apiName := range requested { + endpoint, ok := apiConfigs.APIs[apiName] + if !ok { + result.addFailure(apiName, fmt.Sprintf("API %s is not configured", apiName)) + continue + } + if !endpoint.Enabled { + result.addFailure(apiName, fmt.Sprintf("API %s is disabled", apiName)) + continue + } + + textLanguage := endpoint.TextLanguage + if textLanguage == "" { + textLanguage = "en" + } + + item, err := repository.GetRepositoryByURL(url, textLanguage) + if err != nil { + result.addFailure(apiName, fmt.Sprintf("failed to get repository (language %s): %v", textLanguage, err)) + continue + } + itemPosted = itemPosted || item.Posted + + if endpoint.SocialifyImage && imageName == "" { + imageName, err = generateRetryImage(item.URL) + if err != nil { + result.addFailure(apiName, fmt.Sprintf("failed to prepare image: %v", err)) + continue + } + } + + resp, err := publishItem(apiName, endpoint, *item, imageName) + + switch { + case err != nil: + log.Errorf("%s API error during manual retry: %v", apiName, err) + result.addFailure(apiName, err.Error()) + case resp.Success: + log.Debugf("%s post created successfully during manual retry with language %s!", apiName, textLanguage) + result.Succeeded = append(result.Succeeded, apiName) + result.Outcomes = append(result.Outcomes, RetryOutcome{APIName: apiName, Success: true}) + default: + log.Errorf("%s API request failed during manual retry (status %d): %s", apiName, resp.StatusCode, string(resp.Body)) + result.addFailure(apiName, fmt.Sprintf("API request failed with status %d", resp.StatusCode)) + } + } + + // Marking an unposted item as posted while some connector still failed would + // drop it out of the queue - the very failure this endpoint exists to repair - + // so it is only marked once every requested connector has it. + if !itemPosted && len(result.Succeeded) > 0 && len(result.Failed) == 0 { + if _, err := repository.UpdateRepositoryPosted(url, true); err != nil { + log.Errorf("Failed to update posted status for %s after manual retry: %v", url, err) + } + } + + switch { + case len(result.Succeeded) == 0: + result.Status = 0 + result.Message = fmt.Sprintf("Manual retry: nothing sent for %s. Errors: %s", url, result.errorSummary()) + case len(result.Failed) > 0: + result.Status = 2 + result.Message = fmt.Sprintf("Manual retry: %s sent to: %s. Failed: %s. Errors: %s", + url, strings.Join(result.Succeeded, ", "), strings.Join(result.Failed, ", "), result.errorSummary()) + default: + result.Status = 1 + result.Message = fmt.Sprintf("Manual retry: %s sent to: %s", url, strings.Join(result.Succeeded, ", ")) + } + + // Recorded under the message job so the dashboard's existing filters show it. + // No Pushover notification: a manual retry is already being watched by whoever + // triggered it. + details := &models.MessageRunDetails{ + URL: url, + Sent: result.Succeeded, + Failed: result.Failed, + Manual: true, + } + if err := st.LogCronExecutionDetails("message", result.Status, result.Message, details); err != nil { + log.Errorf("Failed to log manual retry execution: %v", err) + } + + return result, nil +} + +func (r *RetryResult) addFailure(apiName, message string) { + r.Failed = append(r.Failed, apiName) + r.Outcomes = append(r.Outcomes, RetryOutcome{APIName: apiName, Success: false, Error: message}) +} + +func (r *RetryResult) errorSummary() string { + messages := make([]string, 0, len(r.Outcomes)) + for _, outcome := range r.Outcomes { + if !outcome.Success { + messages = append(messages, fmt.Sprintf("%s: %s", outcome.APIName, outcome.Error)) + } + } + return strings.Join(messages, "; ") +} + +func normalizeAPINames(apiNames []string) ([]string, error) { + seen := make(map[string]bool, len(apiNames)) + normalized := make([]string, 0, len(apiNames)) + + for _, name := range apiNames { + name = strings.TrimSpace(name) + if name == "" || seen[name] { + continue + } + seen[name] = true + normalized = append(normalized, name) + } + + if len(normalized) == 0 { + return nil, fmt.Errorf("%w: at least one API name is required", ErrInvalidRetryRequest) + } + + return normalized, nil +} + +// generateRetryImage writes the image into a subdirectory of the image root, so +// the cron's cleanup - which only touches files - cannot delete it while the +// connector is still fetching it. +func generateRetryImage(repoURL string) (string, error) { + if err := os.MkdirAll(retryImageDir, 0o777); err != nil { + return "", fmt.Errorf("failed to create retry image directory: %w", err) + } + + usernameRepo := strings.TrimPrefix(repoURL, "https://github.com/") + imageName := fmt.Sprintf("%s/retry_%d.png", retryImageDir, time.Now().UnixNano()) + + if err := socialify.SocialifyWithConfig(usernameRepo, imageName, retrySocialifyConfig); err != nil { + log.Errorf("Socialify failed during manual retry: %v", err) + if err := utils.CopyFile("./assets/banner.jpg", imageName); err != nil { + // A failed generation can still have left a partial file behind. + if removeErr := os.Remove(imageName); removeErr != nil && !os.IsNotExist(removeErr) { + log.Errorf("Failed to remove partial retry image %s: %v", imageName, removeErr) + } + return "", fmt.Errorf("failed to copy fallback banner file: %w", err) + } + } + + return imageName, nil +} diff --git a/internal/schedule/message-publish_test.go b/internal/schedule/message-publish_test.go new file mode 100644 index 0000000..051cc0d --- /dev/null +++ b/internal/schedule/message-publish_test.go @@ -0,0 +1,379 @@ +package schedule + +import ( + "content-maestro/internal/api" + "content-maestro/internal/models" + "content-maestro/internal/store" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +// retryStore is a StoreInterface stub that only serves API configs and captures +// what the retry recorded in the cron history. +type retryStore struct { + configs []models.APIConfigModel + + loggedName string + loggedStatus int + loggedOutput string + loggedDetails *models.MessageRunDetails + logCalls int +} + +func (s *retryStore) GetAllAPIConfigs() ([]models.APIConfigModel, error) { + return s.configs, nil +} + +func (s *retryStore) LogCronExecutionDetails(name string, status int, output string, details *models.MessageRunDetails) error { + s.logCalls++ + s.loggedName = name + s.loggedStatus = status + s.loggedOutput = output + s.loggedDetails = details + return nil +} + +func (s *retryStore) LogCronExecution(name string, status int, output string) error { + return s.LogCronExecutionDetails(name, status, output, nil) +} + +func (s *retryStore) Close() error { return nil } +func (s *retryStore) InitializeDefaultSettings() error { return nil } +func (s *retryStore) GetCronSetting(string) (*models.CronSetting, error) { + return nil, errors.New("not implemented") +} +func (s *retryStore) GetAllCronSettings() ([]models.CronSetting, error) { + return nil, errors.New("not implemented") +} +func (s *retryStore) UpdateCronSetting(string, string, bool) (*models.CronSetting, error) { + return nil, errors.New("not implemented") +} +func (s *retryStore) GetCronHistoryCount(string, *int, *time.Time, *time.Time) (int, error) { + return 0, errors.New("not implemented") +} +func (s *retryStore) GetCronHistory(string, *int, int, int, string, *time.Time, *time.Time) ([]models.CronHistory, error) { + return nil, errors.New("not implemented") +} +func (s *retryStore) GetCollectSettings() (*store.CollectSettings, error) { + return nil, errors.New("not implemented") +} +func (s *retryStore) UpdateCollectSettings(*store.CollectSettings) error { + return errors.New("not implemented") +} +func (s *retryStore) GetPromptSettings() (*models.PromptSettings, error) { + return nil, errors.New("not implemented") +} +func (s *retryStore) UpdatePromptSettings(*models.UpdatePromptSettingsRequest) error { + return errors.New("not implemented") +} +func (s *retryStore) GetAPIConfig(string) (*models.APIConfigModel, error) { + return nil, errors.New("not implemented") +} +func (s *retryStore) CreateAPIConfig(*models.CreateAPIConfigRequest) (*models.APIConfigModel, error) { + return nil, errors.New("not implemented") +} +func (s *retryStore) UpdateAPIConfig(string, *models.UpdateAPIConfigRequest) (*models.APIConfigModel, error) { + return nil, errors.New("not implemented") +} +func (s *retryStore) DeleteAPIConfig(string) error { return errors.New("not implemented") } + +var _ store.StoreInterface = (*retryStore)(nil) + +const retryTestURL = "https://github.com/resemble-ai/chatterbox" + +// alchemistStub answers get-repository for a single known repository. +func alchemistStub(t *testing.T, posted bool, calls *[]string) *httptest.Server { + t.Helper() + + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var body struct { + URL string `json:"url"` + Posted *bool `json:"posted"` + TextLanguage string `json:"text_language"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + t.Errorf("failed to decode alchemist request: %v", err) + } + if calls != nil { + *calls = append(*calls, body.TextLanguage) + } + + url := body.URL + if url == "" { + // The "latest published" lookup, used when no url is supplied. + url = retryTestURL + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "status": "ok", + "data": map[string]any{ + "items": []map[string]any{{ + "id": 1327, + "posted": posted, + "url": url, + "text": "text for " + body.TextLanguage, + }}, + }, + }) + })) +} + +// withRepositoryEndpoints points the repository package at the stub server. +func withRepositoryEndpoints(t *testing.T, alchemistURL string) { + t.Helper() + + t.Setenv("CONTENT_ALCHEMIST_URL", alchemistURL) + t.Setenv("CONTENT_ALCHEMIST_BEARER", "test-token") +} + +func TestRetryMessagePostSendsOnlyRequestedAPIs(t *testing.T) { + var connectorBodies []map[string]any + connector := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var body map[string]any + json.NewDecoder(r.Body).Decode(&body) + connectorBodies = append(connectorBodies, body) + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{"status":"ok"}`)) + })) + defer connector.Close() + + var languages []string + alchemist := alchemistStub(t, true, &languages) + defer alchemist.Close() + withRepositoryEndpoints(t, alchemist.URL) + + st := &retryStore{configs: []models.APIConfigModel{ + { + Name: "threads", URL: connector.URL, Method: http.MethodPost, + ContentType: "json", SuccessCode: http.StatusOK, Enabled: true, + TextLanguage: "en", + }, + { + Name: "bluesky", URL: connector.URL, Method: http.MethodPost, + ContentType: "json", SuccessCode: http.StatusOK, Enabled: true, + TextLanguage: "uk", + }, + }} + if err := api.LoadAPIConfigs(st); err != nil { + t.Fatalf("LoadAPIConfigs() error = %v", err) + } + + result, err := RetryMessagePost(st, []string{"threads", "threads", " "}, retryTestURL) + if err != nil { + t.Fatalf("RetryMessagePost() error = %v", err) + } + + if len(connectorBodies) != 1 { + t.Fatalf("connector received %d requests, want 1", len(connectorBodies)) + } + if got := connectorBodies[0]["url"]; got != retryTestURL { + t.Errorf("published url = %v, want %v", got, retryTestURL) + } + if got := connectorBodies[0]["text"]; got != "text for en" { + t.Errorf("published text = %v, want the English text", got) + } + if len(languages) != 1 || languages[0] != "en" { + t.Errorf("alchemist was asked for languages %v, want [en]", languages) + } + + if result.Status != 1 { + t.Errorf("status = %d, want 1", result.Status) + } + if len(result.Succeeded) != 1 || result.Succeeded[0] != "threads" { + t.Errorf("succeeded = %v, want [threads]", result.Succeeded) + } + if len(result.Failed) != 0 { + t.Errorf("failed = %v, want none", result.Failed) + } + + if st.logCalls != 1 { + t.Fatalf("cron history writes = %d, want 1", st.logCalls) + } + if st.loggedName != "message" { + t.Errorf("history name = %q, want %q", st.loggedName, "message") + } + if !strings.HasPrefix(st.loggedOutput, "Manual retry:") { + t.Errorf("history output = %q, want a Manual retry prefix", st.loggedOutput) + } + if st.loggedDetails == nil || !st.loggedDetails.Manual { + t.Fatalf("history details = %+v, want manual details", st.loggedDetails) + } + if st.loggedDetails.URL != retryTestURL { + t.Errorf("history details url = %q, want %q", st.loggedDetails.URL, retryTestURL) + } +} + +func TestRetryMessagePostReportsPerAPIFailures(t *testing.T) { + alchemist := alchemistStub(t, true, nil) + defer alchemist.Close() + withRepositoryEndpoints(t, alchemist.URL) + + st := &retryStore{configs: []models.APIConfigModel{ + { + Name: "bluesky", URL: "http://127.0.0.1:1", Method: http.MethodPost, + ContentType: "json", SuccessCode: http.StatusOK, Enabled: false, + TextLanguage: "en", + }, + }} + if err := api.LoadAPIConfigs(st); err != nil { + t.Fatalf("LoadAPIConfigs() error = %v", err) + } + + result, err := RetryMessagePost(st, []string{"bluesky", "telegram"}, retryTestURL) + if err != nil { + t.Fatalf("RetryMessagePost() error = %v", err) + } + + if result.Status != 0 { + t.Errorf("status = %d, want 0", result.Status) + } + if len(result.Failed) != 2 { + t.Fatalf("failed = %v, want both APIs", result.Failed) + } + + reasons := map[string]string{} + for _, outcome := range result.Outcomes { + reasons[outcome.APIName] = outcome.Error + } + if !strings.Contains(reasons["bluesky"], "disabled") { + t.Errorf("bluesky outcome = %q, want a disabled reason", reasons["bluesky"]) + } + if !strings.Contains(reasons["telegram"], "not configured") { + t.Errorf("telegram outcome = %q, want a not-configured reason", reasons["telegram"]) + } +} + +func TestRetryMessagePostRejectsEmptyAPIList(t *testing.T) { + st := &retryStore{} + if err := api.LoadAPIConfigs(st); err != nil { + t.Fatalf("LoadAPIConfigs() error = %v", err) + } + + _, err := RetryMessagePost(st, []string{" ", ""}, retryTestURL) + if !errors.Is(err, ErrInvalidRetryRequest) { + t.Fatalf("error = %v, want ErrInvalidRetryRequest", err) + } + if st.logCalls != 0 { + t.Errorf("cron history writes = %d, want none for a rejected request", st.logCalls) + } +} + +// An unposted item must only leave the queue once every requested connector has +// it: marking it posted after a partial retry re-creates the failure this +// endpoint exists to repair. +func TestRetryMessagePostMarksPostedOnlyWhenComplete(t *testing.T) { + tests := []struct { + name string + apis []string + wantPosted bool + }{ + {name: "every requested integration succeeds", apis: []string{"threads"}, wantPosted: true}, + {name: "one integration still fails", apis: []string{"threads", "mastodon"}, wantPosted: false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + connector := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + w.Write([]byte(`{"status":"ok"}`)) + })) + defer connector.Close() + + var postedPatches int + alchemist := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodPatch { + postedPatches++ + w.Write([]byte(`{"status":"ok","message":"updated"}`)) + return + } + + var body struct { + URL string `json:"url"` + TextLanguage string `json:"text_language"` + } + json.NewDecoder(r.Body).Decode(&body) + + url := body.URL + if url == "" { + url = retryTestURL + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "status": "ok", + "data": map[string]any{ + "items": []map[string]any{{ + "id": 1327, "posted": false, "url": url, "text": "text", + }}, + }, + }) + })) + defer alchemist.Close() + withRepositoryEndpoints(t, alchemist.URL) + + st := &retryStore{configs: []models.APIConfigModel{{ + Name: "threads", URL: connector.URL, Method: http.MethodPost, + ContentType: "json", SuccessCode: http.StatusOK, Enabled: true, + TextLanguage: "en", + }}} + if err := api.LoadAPIConfigs(st); err != nil { + t.Fatalf("LoadAPIConfigs() error = %v", err) + } + + if _, err := RetryMessagePost(st, tt.apis, retryTestURL); err != nil { + t.Fatalf("RetryMessagePost() error = %v", err) + } + + if tt.wantPosted && postedPatches != 1 { + t.Errorf("update-posted calls = %d, want 1", postedPatches) + } + if !tt.wantPosted && postedPatches != 0 { + t.Errorf("update-posted calls = %d, want none while a connector still failed", postedPatches) + } + }) + } +} + +// A retry keeps its image in a subdirectory of the image root, so the served +// path has to carry that subdirectory or the connector fetches a 404. +func TestImageURLPath(t *testing.T) { + tests := []struct { + name string + image string + want string + }{ + {name: "cron image at the root", image: imageDir + "/image_123.png", want: "image_123.png"}, + {name: "retry image in a subdirectory", image: retryImageDir + "/retry_123.png", want: "retry/retry_123.png"}, + {name: "path outside the image root falls back to the file name", image: "/var/tmp/other.png", want: "other.png"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := imageURLPath(tt.image); got != tt.want { + t.Errorf("imageURLPath(%q) = %q, want %q", tt.image, got, tt.want) + } + }) + } +} + +func TestNormalizeAPINames(t *testing.T) { + got, err := normalizeAPINames([]string{" threads ", "threads", "bluesky", ""}) + if err != nil { + t.Fatalf("normalizeAPINames() error = %v", err) + } + + want := []string{"threads", "bluesky"} + if len(got) != len(want) { + t.Fatalf("normalizeAPINames() = %v, want %v", got, want) + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("normalizeAPINames() = %v, want %v", got, want) + } + } +} diff --git a/internal/schedule/message-schedule.go b/internal/schedule/message-schedule.go index 5f7a03b..2feb0dd 100644 --- a/internal/schedule/message-schedule.go +++ b/internal/schedule/message-schedule.go @@ -2,14 +2,13 @@ package schedule import ( "content-maestro/internal/api" + "content-maestro/internal/models" "content-maestro/internal/notification" "content-maestro/internal/repository" "content-maestro/internal/socialify" "content-maestro/internal/store" "content-maestro/internal/utils" "fmt" - "os" - "path/filepath" "strings" "time" @@ -22,18 +21,38 @@ func MessageJob(s *gocron.Scheduler, store store.StoreInterface) { var status int var logMessage string + var successfulAPIs []string + var failedAPIs []string + var errorMessages []string + var updatedURL string + + // Assembled at exit rather than at the end of the publishing loop, so an + // early return - or a panic - still records which item the run consumed and + // which connectors missed it. That is the data a manual retry needs, and it + // matters most exactly when the run died mid-publish. + runDetails := func() *models.MessageRunDetails { + if updatedURL == "" && len(successfulAPIs) == 0 && len(failedAPIs) == 0 { + return nil + } + return &models.MessageRunDetails{ + URL: updatedURL, + Sent: successfulAPIs, + Failed: failedAPIs, + } + } + defer func() { if r := recover(); r != nil { panicMessage := fmt.Sprintf("Panic occurred: %v. %s", r, logMessage) log.Error("Message job panic: %v", r) - if err := store.LogCronExecution("message", 0, panicMessage); err != nil { + if err := store.LogCronExecutionDetails("message", 0, panicMessage, runDetails()); err != nil { log.Error("Failed to log panic execution: %v", err) } notification.NotifyCronResult("message", 0, panicMessage) panic(r) } - if err := store.LogCronExecution("message", status, logMessage); err != nil { + if err := store.LogCronExecutionDetails("message", status, logMessage, runDetails()); err != nil { log.Error("Failed to log cron execution: %v", err) } notification.NotifyCronResult("message", status, logMessage) @@ -83,7 +102,7 @@ func MessageJob(s *gocron.Scheduler, store store.StoreInterface) { username_repo := strings.TrimPrefix(item.URL, "https://github.com/") timestamp := time.Now().UnixNano() imageFilename := fmt.Sprintf("image_%d.png", timestamp) - image_name = fmt.Sprintf("./tmp/gh_project_img/%s", imageFilename) + image_name = fmt.Sprintf("%s/%s", imageDir, imageFilename) err = socialify.Socialify(username_repo, image_name) if err != nil { @@ -100,11 +119,6 @@ func MessageJob(s *gocron.Scheduler, store store.StoreInterface) { } } - var successfulAPIs []string - var failedAPIs []string - var errorMessages []string - var updatedURL string - for apiName, endpoint := range apiConfigs.APIs { if !endpoint.Enabled { continue @@ -132,6 +146,13 @@ func MessageJob(s *gocron.Scheduler, store store.StoreInterface) { item := repo.Data.Items[0] + // Repositories whose URL no longer resolves are dropped and the next + // candidate is fetched. The loop reports its outcome through this flag: + // reading the fetch response afterwards would dereference nil once a + // re-fetch fails, which crashes the job and, through the deferred + // re-panic, the whole process. + itemAvailable := true + for { statusCode, err := repository.ValidateRepositoryURL(item.URL) if err != nil { @@ -150,25 +171,27 @@ func MessageJob(s *gocron.Scheduler, store store.StoreInterface) { log.Error("Error deleting repository %s: %v", item.URL, err) } - repo, err = repository.GetRepository(1, false, "ASC", "publication_queue", textLanguage) + nextRepo, err := repository.GetRepository(1, false, "ASC", "publication_queue", textLanguage) if err != nil { log.Error("Error getting next repository for %s API: %v", apiName, err) failedAPIs = append(failedAPIs, apiName) errorMessages = append(errorMessages, fmt.Sprintf("%s API error: failed to get next repository: %v", apiName, err)) + itemAvailable = false break } - if len(repo.Data.Items) == 0 { + if len(nextRepo.Data.Items) == 0 { log.Debugf("No more valid repositories available for %s API", apiName) failedAPIs = append(failedAPIs, apiName) errorMessages = append(errorMessages, fmt.Sprintf("%s API error: no valid repositories available", apiName)) + itemAvailable = false break } - item = repo.Data.Items[0] + item = nextRepo.Data.Items[0] } - if len(repo.Data.Items) == 0 { + if !itemAvailable { continue } @@ -176,48 +199,7 @@ func MessageJob(s *gocron.Scheduler, store store.StoreInterface) { updatedURL = item.URL } - var req api.RequestConfig - - commonFields := map[string]string{ - "text": item.Text, - "url": item.URL, - } - - switch strings.ToLower(endpoint.ContentType) { - case "multipart": - req = api.RequestConfig{ - APIName: apiName, - FormFields: commonFields, - } - - if endpoint.SocialifyImage && image_name != "" { - req.FileFields = map[string]string{ - "image": image_name, - } - } - case "json": - req = api.RequestConfig{ - APIName: apiName, - JSONBody: map[string]any{"text": item.Text, "url": item.URL}, - } - - if endpoint.SocialifyImage && image_name != "" { - publicURL := os.Getenv("PUBLIC_URL") - if publicURL != "" { - imageURL := fmt.Sprintf("%s/images/%s", publicURL, filepath.Base(image_name)) - req.JSONBody["image_url"] = imageURL - } else { - log.Error("PUBLIC_URL not set, cannot generate image_url for API %s", apiName) - } - } - default: - req = api.RequestConfig{ - APIName: apiName, - JSONBody: map[string]any{"text": item.Text, "url": item.URL}, - } - } - - resp, err := api.ExecuteRequest(req) + resp, err := publishItem(apiName, endpoint, item, image_name) if err != nil { log.Errorf("%s API error: %v", apiName, err) failedAPIs = append(failedAPIs, apiName) @@ -241,7 +223,7 @@ func MessageJob(s *gocron.Scheduler, store store.StoreInterface) { } } - err := utils.RemoveAllFilesInFolder("./tmp/gh_project_img") + err := utils.RemoveAllFilesInFolder(imageDir) if err != nil { log.Error(err) status = 0 diff --git a/internal/schedule/message-schedule_test.go b/internal/schedule/message-schedule_test.go new file mode 100644 index 0000000..2421b77 --- /dev/null +++ b/internal/schedule/message-schedule_test.go @@ -0,0 +1,180 @@ +package schedule + +import ( + "content-maestro/internal/api" + "content-maestro/internal/models" + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "strings" + "testing" +) + +// withMessageJobWorkdir gives the job an empty working directory containing the +// image folder it cleans up, so a test never writes into the repository. +func withMessageJobWorkdir(t *testing.T) { + t.Helper() + + t.Chdir(t.TempDir()) + if err := os.MkdirAll(imageDir, 0o777); err != nil { + t.Fatalf("failed to create image directory: %v", err) + } +} + +// queueStub serves the publication queue and can start rejecting requests after +// a given number of calls, which is how a re-fetch fails mid-run. +type queueStub struct { + calls int + rejectAfter int + repositoryPath string + repositoryURL string + validationCode int +} + +func (q *queueStub) handler(t *testing.T) http.HandlerFunc { + t.Helper() + + return func(w http.ResponseWriter, r *http.Request) { + // The repository URL validation performs a HEAD request; answer it with the + // configured status so the revalidation loop can be driven from a test. + if r.Method == http.MethodHead { + w.WriteHeader(q.validationCode) + return + } + + if r.Method == http.MethodDelete || r.Method == http.MethodPatch { + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"status":"ok","message":"done"}`)) + return + } + + q.calls++ + if q.rejectAfter > 0 && q.calls > q.rejectAfter { + w.WriteHeader(http.StatusTooManyRequests) + w.Write([]byte("Too Many Requests")) + return + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "status": "ok", + "data": map[string]any{ + "items": []map[string]any{{ + "id": 1, "posted": false, "url": q.repositoryURL, "text": "text", + }}, + }, + }) + } +} + +// A failed re-fetch inside the revalidation loop used to leave the response nil +// and crash the job on the next read, which the deferred re-panic turned into a +// process exit. +func TestMessageJobSurvivesRejectedRefetch(t *testing.T) { + stub := &queueStub{ + rejectAfter: 1, + repositoryPath: "/dead/repo", + validationCode: http.StatusNotFound, + } + server := httptest.NewServer(stub.handler(t)) + defer server.Close() + stub.repositoryURL = server.URL + stub.repositoryPath + withMessageJobWorkdir(t) + + t.Setenv("CONTENT_ALCHEMIST_URL", server.URL) + t.Setenv("CONTENT_ALCHEMIST_BEARER", "test-token") + + st := &retryStore{configs: []models.APIConfigModel{{ + Name: "threads", URL: server.URL, Method: http.MethodPost, + ContentType: "json", SuccessCode: http.StatusOK, Enabled: true, + TextLanguage: "en", + }}} + if err := api.LoadAPIConfigs(st); err != nil { + t.Fatalf("LoadAPIConfigs() error = %v", err) + } + + defer func() { + if r := recover(); r != nil { + t.Fatalf("MessageJob panicked: %v", r) + } + }() + + MessageJob(nil, st) + + if st.logCalls != 1 { + t.Fatalf("cron history writes = %d, want 1", st.logCalls) + } + if st.loggedStatus != 0 { + t.Errorf("status = %d, want 0", st.loggedStatus) + } + if !strings.Contains(st.loggedOutput, "failed to get next repository") { + t.Errorf("output = %q, want the re-fetch failure reported", st.loggedOutput) + } + // The run consumed no item, so there is nothing to retry and no url to record. + if st.loggedDetails == nil || len(st.loggedDetails.Failed) != 1 || st.loggedDetails.Failed[0] != "threads" { + t.Fatalf("details = %+v, want threads recorded as failed", st.loggedDetails) + } + if st.loggedDetails.URL != "" { + t.Errorf("details url = %q, want empty for a run that published nothing", st.loggedDetails.URL) + } +} + +// A run that cannot even read its configuration has nothing worth recording, so +// consumers can treat "has details" as "has something to retry". +func TestMessageJobRecordsNoDetailsWithoutConfigs(t *testing.T) { + withMessageJobWorkdir(t) + + st := &retryStore{} + if err := api.LoadAPIConfigs(st); err != nil { + t.Fatalf("LoadAPIConfigs() error = %v", err) + } + + MessageJob(nil, st) + + if st.logCalls != 1 { + t.Fatalf("cron history writes = %d, want 1", st.logCalls) + } + if st.loggedDetails != nil { + t.Errorf("details = %+v, want nil", st.loggedDetails) + } +} + +func TestMessageJobRecordsDetailsOnSuccess(t *testing.T) { + stub := &queueStub{repositoryPath: "/live/repo", validationCode: http.StatusOK} + server := httptest.NewServer(stub.handler(t)) + defer server.Close() + repoURL := server.URL + stub.repositoryPath + stub.repositoryURL = repoURL + withMessageJobWorkdir(t) + + t.Setenv("CONTENT_ALCHEMIST_URL", server.URL) + t.Setenv("CONTENT_ALCHEMIST_BEARER", "test-token") + + st := &retryStore{configs: []models.APIConfigModel{{ + Name: "threads", URL: server.URL, Method: http.MethodPost, + ContentType: "json", SuccessCode: http.StatusOK, Enabled: true, + TextLanguage: "en", + }}} + if err := api.LoadAPIConfigs(st); err != nil { + t.Fatalf("LoadAPIConfigs() error = %v", err) + } + + MessageJob(nil, st) + + if st.loggedStatus != 1 { + t.Fatalf("status = %d, want 1 (output: %s)", st.loggedStatus, st.loggedOutput) + } + if st.loggedDetails == nil { + t.Fatal("details = nil, want the published item recorded") + } + if st.loggedDetails.URL != repoURL { + t.Errorf("details url = %q, want %q", st.loggedDetails.URL, repoURL) + } + if len(st.loggedDetails.Sent) != 1 || st.loggedDetails.Sent[0] != "threads" { + t.Errorf("details sent = %v, want [threads]", st.loggedDetails.Sent) + } + if st.loggedDetails.Manual { + t.Error("details manual = true, want false for a scheduled run") + } +} diff --git a/internal/server/api.go b/internal/server/api.go index f7d62f7..e658e82 100644 --- a/internal/server/api.go +++ b/internal/server/api.go @@ -3,9 +3,11 @@ package server import ( apiExecutor "content-maestro/internal/api" "content-maestro/internal/models" + "content-maestro/internal/schedule" "content-maestro/internal/store" "content-maestro/internal/validation" "encoding/json" + "errors" "fmt" "net/http" "strconv" @@ -468,6 +470,44 @@ func (api *CronAPI) DeleteAPIConfig(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(response) } +// RetryMessagePost re-sends an already published repository to the APIs that +// missed it. A partially successful message run marks the item as posted, so the +// cron alone can never recover the connectors that failed. +// +// The response is always 200 once the retry has run: per-API outcomes carry the +// individual failures, which are far more useful than a single status code. +func (api *CronAPI) RetryMessagePost(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodOptions: + return + case http.MethodPost: + default: + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + + var req models.RetryMessageRequest + // The body is a short list of integration names and a url; anything larger is + // not a request this endpoint should read into memory. + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 8*1024)).Decode(&req); err != nil { + http.Error(w, "Invalid request body", http.StatusBadRequest) + return + } + + result, err := schedule.RetryMessagePost(api.store, req.APIs, req.URL) + if err != nil { + if errors.Is(err, schedule.ErrInvalidRetryRequest) { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(result) +} + func (api *CronAPI) HandleAPIConfigs(w http.ResponseWriter, r *http.Request) { switch r.Method { case http.MethodOptions: diff --git a/internal/socialify/socialify.go b/internal/socialify/socialify.go index bc73a22..cf34e8d 100644 --- a/internal/socialify/socialify.go +++ b/internal/socialify/socialify.go @@ -12,7 +12,11 @@ import ( ) var log = logger.NewLogger() -var SocialifyHTTPClient = &http.Client{} + +// A timeout is required, not cosmetic: without one a hung upstream connection +// bounds neither an attempt nor the caller, and the manual retry endpoint answers +// an HTTP request synchronously. +var SocialifyHTTPClient = &http.Client{Timeout: 30 * time.Second} type RetryConfig struct { MaxRetries int @@ -35,10 +39,21 @@ func ResetRetryConfig() { } func Socialify(usernameRepo string, outputPath string) error { + return SocialifyWithConfig(usernameRepo, outputPath, currentConfig) +} + +// SocialifyWithConfig runs the image generation with a caller-supplied retry +// budget. Callers that answer a synchronous request use it to avoid the default +// budget, which can block for minutes, without mutating the shared config. +func SocialifyWithConfig(usernameRepo string, outputPath string, config RetryConfig) error { log.Debug("Starting Socialify image parsing") + if config.MaxRetries < 1 { + config.MaxRetries = 1 + } + var lastErr error - for attempt := 1; attempt <= currentConfig.MaxRetries; attempt++ { + for attempt := 1; attempt <= config.MaxRetries; attempt++ { err := trySocialify(usernameRepo, outputPath) if err == nil { log.Debug("Socialify image parsing finished") @@ -46,13 +61,13 @@ func Socialify(usernameRepo string, outputPath string) error { } lastErr = err - if attempt < currentConfig.MaxRetries { - log.Errorf("Attempt %d failed: %v. Retrying in %s...", attempt, err, currentConfig.RetryInterval) - time.Sleep(currentConfig.RetryInterval) + if attempt < config.MaxRetries { + log.Errorf("Attempt %d failed: %v. Retrying in %s...", attempt, err, config.RetryInterval) + time.Sleep(config.RetryInterval) } } - log.Debugf("All %d attempts failed. Last error: %v", currentConfig.MaxRetries, lastErr) + log.Debugf("All %d attempts failed. Last error: %v", config.MaxRetries, lastErr) return lastErr } diff --git a/internal/store/sqlite_store.go b/internal/store/sqlite_store.go index ca94869..e9238b2 100644 --- a/internal/store/sqlite_store.go +++ b/internal/store/sqlite_store.go @@ -8,8 +8,8 @@ import ( "os" "time" - _ "modernc.org/sqlite" "gopkg.in/yaml.v3" + _ "modernc.org/sqlite" ) const ( @@ -75,7 +75,8 @@ func createTablesIfNotExist(db *sql.DB) error { name TEXT NOT NULL, timestamp DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, status INTEGER NOT NULL, - output TEXT + output TEXT, + details TEXT )`) if err != nil { return fmt.Errorf("failed to create cron_history table: %v", err) @@ -103,6 +104,10 @@ func createTablesIfNotExist(db *sql.DB) error { return fmt.Errorf("failed to migrate cron_history success to status: %v", err) } + if err := migrateCronHistoryDetails(db); err != nil { + return fmt.Errorf("failed to migrate cron_history details: %v", err) + } + _, err = db.Exec(` INSERT OR IGNORE INTO cron_settings (name, schedule, is_active, updated_at) VALUES ('collect', '13 13 * * 6', 0, CURRENT_TIMESTAMP)`) @@ -230,39 +235,59 @@ func migrateCollectSettingsSchema(db *sql.DB) error { } func migrateCronHistorySuccessToStatus(db *sql.DB) error { + columns, err := cronHistoryColumns(db) + if err != nil { + return err + } + + if columns["status"] || !columns["success"] { + return nil + } + + if _, err := db.Exec("ALTER TABLE cron_history RENAME COLUMN success TO status"); err != nil { + return fmt.Errorf("failed to rename success column to status: %v", err) + } + + return nil +} + +func migrateCronHistoryDetails(db *sql.DB) error { + columns, err := cronHistoryColumns(db) + if err != nil { + return err + } + + if columns["details"] { + return nil + } + + if _, err := db.Exec("ALTER TABLE cron_history ADD COLUMN details TEXT"); err != nil { + return fmt.Errorf("failed to add details column: %v", err) + } + + return nil +} + +func cronHistoryColumns(db *sql.DB) (map[string]bool, error) { rows, err := db.Query("PRAGMA table_info(cron_history)") if err != nil { - return fmt.Errorf("failed to query table info: %v", err) + return nil, fmt.Errorf("failed to query table info: %v", err) } defer rows.Close() - hasSuccessColumn := false - hasStatusColumn := false + columns := make(map[string]bool) for rows.Next() { var cid int var name, colType string var notNull, pk int var dfltValue interface{} if err := rows.Scan(&cid, &name, &colType, ¬Null, &dfltValue, &pk); err != nil { - return fmt.Errorf("failed to scan table info: %v", err) - } - if name == "success" { - hasSuccessColumn = true + return nil, fmt.Errorf("failed to scan table info: %v", err) } - if name == "status" { - hasStatusColumn = true - } - } - - if hasStatusColumn || !hasSuccessColumn { - return nil - } - - if _, err := db.Exec("ALTER TABLE cron_history RENAME COLUMN success TO status"); err != nil { - return fmt.Errorf("failed to rename success column to status: %v", err) + columns[name] = true } - return nil + return columns, rows.Err() } func (s *SQLiteStore) Close() error { @@ -371,6 +396,10 @@ func (s *SQLiteStore) InitializeDefaultSettings() error { } func (s *SQLiteStore) LogCronExecution(name string, status int, output string) error { + return s.LogCronExecutionDetails(name, status, output, nil) +} + +func (s *SQLiteStore) LogCronExecutionDetails(name string, status int, output string, details *models.MessageRunDetails) error { if name == "" { return fmt.Errorf("cron job name cannot be empty") } @@ -386,8 +415,17 @@ func (s *SQLiteStore) LogCronExecution(name string, status int, output string) e timestamp := time.Now() - query := "INSERT INTO cron_history (name, timestamp, status, output) VALUES (?, ?, ?, ?)" - _, err := s.db.Exec(query, name, timestamp, status, output) + var encodedDetails any + if details != nil { + encoded, err := json.Marshal(details) + if err != nil { + return fmt.Errorf("failed to encode cron execution details: %v", err) + } + encodedDetails = string(encoded) + } + + query := "INSERT INTO cron_history (name, timestamp, status, output, details) VALUES (?, ?, ?, ?, ?)" + _, err := s.db.Exec(query, name, timestamp, status, output, encodedDetails) if err != nil { fmt.Printf("Failed to log cron execution to database: %v\n", err) fmt.Printf("Attempted to log: name=%s, status=%d, timestamp=%v, output_length=%d\n", @@ -431,7 +469,7 @@ func (s *SQLiteStore) GetCronHistoryCount(name string, status *int, startDate, e } func (s *SQLiteStore) GetCronHistory(name string, status *int, offset, limit int, sortOrder string, startDate, endDate *time.Time) ([]models.CronHistory, error) { - query := "SELECT name, timestamp, status, output FROM cron_history WHERE 1=1" + query := "SELECT name, timestamp, status, output, details FROM cron_history WHERE 1=1" args := []any{} if name != "" { @@ -469,9 +507,20 @@ func (s *SQLiteStore) GetCronHistory(name string, status *int, offset, limit int var history []models.CronHistory for rows.Next() { var h models.CronHistory - if err := rows.Scan(&h.Name, &h.Timestamp, &h.Success, &h.Output); err != nil { + // details is NULL for every run recorded before the column existed. + var details sql.NullString + if err := rows.Scan(&h.Name, &h.Timestamp, &h.Success, &h.Output, &details); err != nil { return nil, fmt.Errorf("failed to scan cron history: %v", err) } + if details.Valid && details.String != "" { + var parsed models.MessageRunDetails + if err := json.Unmarshal([]byte(details.String), &parsed); err != nil { + // A malformed row must not take the whole history down. + fmt.Printf("Failed to decode cron history details for %s: %v\n", h.Name, err) + } else { + h.Details = &parsed + } + } history = append(history, h) } return history, nil diff --git a/internal/store/sqlite_store_test.go b/internal/store/sqlite_store_test.go index 98760cf..e9b061b 100644 --- a/internal/store/sqlite_store_test.go +++ b/internal/store/sqlite_store_test.go @@ -132,6 +132,31 @@ func TestSQLiteStore_LogCronExecution(t *testing.T) { assert.Equal(t, "Test output", history[0].Output) } +func TestSQLiteStore_LogCronExecutionDetails(t *testing.T) { + store := setupTestStore(t) + defer store.Close() + + details := &models.MessageRunDetails{ + URL: "https://github.com/resemble-ai/chatterbox", + Sent: []string{"bluesky"}, + Failed: []string{"threads"}, + Manual: true, + } + + require.NoError(t, store.LogCronExecutionDetails("message", 2, "Manual retry: partial", details)) + // A run logged without details must read back without them, which is also how + // every row written before the column existed behaves. + require.NoError(t, store.LogCronExecution("message", 1, "Scheduled run")) + + history, err := store.GetCronHistory("message", nil, 0, 10, "asc", nil, nil) + require.NoError(t, err) + require.Len(t, history, 2) + + require.NotNil(t, history[0].Details) + assert.Equal(t, *details, *history[0].Details) + assert.Nil(t, history[1].Details) +} + func TestSQLiteStore_LogCronExecution_EmptyName(t *testing.T) { store := setupTestStore(t) defer store.Close() diff --git a/internal/store/store_interface.go b/internal/store/store_interface.go index 1850638..6e76146 100644 --- a/internal/store/store_interface.go +++ b/internal/store/store_interface.go @@ -12,6 +12,7 @@ type StoreInterface interface { GetAllCronSettings() ([]models.CronSetting, error) UpdateCronSetting(name string, schedule string, isActive bool) (*models.CronSetting, error) LogCronExecution(name string, status int, output string) error + LogCronExecutionDetails(name string, status int, output string, details *models.MessageRunDetails) error GetCronHistoryCount(name string, status *int, startDate, endDate *time.Time) (int, error) GetCronHistory(name string, status *int, offset, limit int, sortOrder string, startDate, endDate *time.Time) ([]models.CronHistory, error) GetCollectSettings() (*CollectSettings, error) diff --git a/internal/utils/remove_files.go b/internal/utils/remove_files.go index 195aecf..11607d9 100644 --- a/internal/utils/remove_files.go +++ b/internal/utils/remove_files.go @@ -19,8 +19,16 @@ func RemoveAllFilesInFolder(dir string) error { } for _, file := range files { - err = remove(file) - if err != nil { + // Subdirectories are owned by whoever created them - a manual retry keeps + // its images in one so this cleanup cannot delete a file that is still + // being uploaded. + if info, err := os.Stat(file); err == nil && info.IsDir() { + continue + } + + // A file that another writer removed in the meantime is not a failure: + // treating it as one used to turn a successful run into a reported one. + if err := remove(file); err != nil && !os.IsNotExist(err) { return err } } diff --git a/internal/utils/remove_files_test.go b/internal/utils/remove_files_test.go index 1bec9ea..e703fd3 100644 --- a/internal/utils/remove_files_test.go +++ b/internal/utils/remove_files_test.go @@ -106,6 +106,48 @@ func TestRemoveAllFilesInFolder_Success(t *testing.T) { } } +// The image folder has two writers: the publishing cron and a manual retry. The +// cleanup must leave the retry's subdirectory alone and must not treat a file +// that vanished under it as a failure - doing so used to turn a successful run +// into a reported failure. +func TestRemoveAllFilesInFolder_ToleratesConcurrentWriters(t *testing.T) { + testDir := t.TempDir() + + subDir := filepath.Join(testDir, "retry") + if err := os.MkdirAll(subDir, 0o777); err != nil { + t.Fatalf("failed to create subdirectory: %v", err) + } + keep := filepath.Join(subDir, "retry_1.png") + if err := os.WriteFile(keep, []byte("image"), 0o644); err != nil { + t.Fatalf("failed to write file in subdirectory: %v", err) + } + + vanishing := filepath.Join(testDir, "image_1.png") + if err := os.WriteFile(vanishing, []byte("image"), 0o644); err != nil { + t.Fatalf("failed to write file: %v", err) + } + + // Simulate the other writer deleting the file first. + remove = func(name string) error { + if err := os.Remove(name); err != nil { + return err + } + return os.Remove(name) + } + defer func() { remove = os.Remove }() + + if err := RemoveAllFilesInFolder(testDir); err != nil { + t.Fatalf("RemoveAllFilesInFolder() error = %v, want nil", err) + } + + if _, err := os.Stat(keep); err != nil { + t.Errorf("file in subdirectory was removed: %v", err) + } + if _, err := os.Stat(vanishing); !os.IsNotExist(err) { + t.Errorf("file at the root was not removed: %v", err) + } +} + func TestRemoveAllFilesInFolder_FailedRemove(t *testing.T) { testDir := t.TempDir() fp := filepath.Join(testDir, "locked")