From 7b4f2e612547454b1819182dbee3440422d2bbfd Mon Sep 17 00:00:00 2001 From: sigmanor Date: Mon, 17 Aug 2026 20:11:30 +0300 Subject: [PATCH 1/6] fix(repository): treat a rejected content-alchemist response as an error A non-2xx answer was decoded as an ordinary payload, which left Items empty and made callers report "no items available". A 400 from content-alchemist therefore surfaced as an empty publication queue: the message cron logged "no items for language en" while the real cause was an invalid-language rejection. Rejections now return an error carrying the message content-alchemist sent, with a fallback to the raw body for the plain-text auth and rate-limit replies. The fallback on "no text available for language" is removed - content-alchemist has not sent that message since it started falling back to an available language. Endpoints and the bearer header are resolved per call instead of at import time, so they no longer depend on when the environment was populated, and request bodies are built with encoding/json rather than string interpolation. --- internal/repository/endpoints.go | 60 ++++++++ internal/repository/get_repository.go | 131 ++++++++++++++---- internal/repository/get_repository_test.go | 130 ++++++++++++++--- internal/repository/update_repository.go | 49 ++----- internal/repository/update_repository_test.go | 78 ++++------- internal/repository/validate_repository.go | 19 +-- .../repository/validate_repository_test.go | 19 +-- 7 files changed, 326 insertions(+), 160 deletions(-) create mode 100644 internal/repository/endpoints.go diff --git a/internal/repository/endpoints.go b/internal/repository/endpoints.go new file mode 100644 index 0000000..c23ed79 --- /dev/null +++ b/internal/repository/endpoints.go @@ -0,0 +1,60 @@ +package repository + +import ( + "net/http" + "os" + "strconv" + "strings" + "sync" + "time" +) + +var ( + client *http.Client + clientOnce 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() { + clientOnce.Do(func() { + client = &http.Client{ + Timeout: getContentAlchemistTimeout(), + } + }) +} + +// 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..179b6c0 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,7 +120,7 @@ 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) @@ -84,10 +129,48 @@ func makeRepositoryRequest(limit int, posted bool, sort_order, sort_by, textLang } 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..460829b 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,7 +32,7 @@ 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) 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..f0e14a4 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 { @@ -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,7 +49,7 @@ func DeleteRepository(url string) (bool, error) { "Accept": {"*/*"}, "Connection": {"keep-alive"}, "Content-Type": {"application/json"}, - "Authorization": {bearerToken}, + "Authorization": {authorizationHeader()}, } resp, err := client.Do(req) 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 { From bceee121c7366f2c9311e86768b4e0a5c658b7b1 Mon Sep 17 00:00:00 2001 From: sigmanor Date: Mon, 17 Aug 2026 20:11:31 +0300 Subject: [PATCH 2/6] feat(history): record which item a message run published and where it landed The cron history only held free-form output, so nothing could tell which repository a run published or which integrations missed it. That is exactly what is needed to finish a partial publication by hand. Adds a details column holding url, sent, failed and manual, exposed through GetCronHistory. LogCronExecution keeps its signature and delegates with no details, so the collect job is unchanged. The column is nullable: runs recorded before it existed read back without details. --- internal/models/cron_history.go | 19 ++++-- internal/store/sqlite_store.go | 99 +++++++++++++++++++++++-------- internal/store/store_interface.go | 1 + 3 files changed, 90 insertions(+), 29 deletions(-) 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/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/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) From ec12727cf4f8bf45eb52d8c2fb080c04fcd22a4a Mon Sep 17 00:00:00 2001 From: sigmanor Date: Mon, 17 Aug 2026 20:11:31 +0300 Subject: [PATCH 3/6] feat(schedule): re-send one item to the integrations that missed it A message run marks its repository as posted as soon as any integration succeeds, so the item leaves the publication queue and the connectors that failed can never recover it on a later run. RetryMessagePost publishes one repository to the named integrations, fetching the text per integration in its configured language and reporting a per-API outcome. When no url is given it uses the most recently published repository, which is the item a partial run just consumed. The request building shared with MessageJob is extracted into publishItem so both paths cannot drift. A retry is recorded in the cron history as a manual message run and sends no Pushover notification - whoever triggered it is already watching. It removes only its own image instead of clearing the shared directory, and it uses a short socialify retry budget via SocialifyWithConfig so a synchronous request cannot block for minutes. --- internal/schedule/message-publish.go | 263 ++++++++++++++++++++ internal/schedule/message-publish_test.go | 282 ++++++++++++++++++++++ internal/schedule/message-schedule.go | 57 +---- internal/socialify/socialify.go | 21 +- 4 files changed, 572 insertions(+), 51 deletions(-) create mode 100644 internal/schedule/message-publish.go create mode 100644 internal/schedule/message-publish_test.go diff --git a/internal/schedule/message-publish.go b/internal/schedule/message-publish.go new file mode 100644 index 0000000..9ba9170 --- /dev/null +++ b/internal/schedule/message-publish.go @@ -0,0 +1,263 @@ +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" + "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, +} + +// 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, filepath.Base(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, which is the +// item a partial run has just consumed. +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 + } + + // 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} + + 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 + + imageName := "" + if endpoint.SocialifyImage { + 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) + + // Only this retry's own file is removed: the cron shares the directory and + // serves its image over /images/ while its own request is in flight. + if imageName != "" { + if removeErr := os.Remove(imageName); removeErr != nil && !os.IsNotExist(removeErr) { + log.Errorf("Failed to remove retry image %s: %v", imageName, removeErr) + } + } + + 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)) + } + } + + if len(result.Succeeded) > 0 && !itemPosted { + 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 +} + +func generateRetryImage(repoURL string) (string, error) { + usernameRepo := strings.TrimPrefix(repoURL, "https://github.com/") + imageName := fmt.Sprintf("./tmp/gh_project_img/retry_%d.png", 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 { + 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..4595e6b --- /dev/null +++ b/internal/schedule/message-publish_test.go @@ -0,0 +1,282 @@ +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) + } +} + +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..553431f 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" @@ -21,19 +20,22 @@ func MessageJob(s *gocron.Scheduler, store store.StoreInterface) { var status int var logMessage string + // Recorded alongside the human-readable output so a partial run can be + // re-sent to the connectors that missed the item. + details := &models.MessageRunDetails{} 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, details); 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, details); err != nil { log.Error("Failed to log cron execution: %v", err) } notification.NotifyCronResult("message", status, logMessage) @@ -176,48 +178,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) @@ -232,6 +193,10 @@ func MessageJob(s *gocron.Scheduler, store store.StoreInterface) { } } + details.URL = updatedURL + details.Sent = successfulAPIs + details.Failed = failedAPIs + if len(successfulAPIs) > 0 && updatedURL != "" { if _, err := repository.UpdateRepositoryPosted(updatedURL, true); err != nil { log.Error("Error updating repository posted status: %v", err) diff --git a/internal/socialify/socialify.go b/internal/socialify/socialify.go index bc73a22..e068f51 100644 --- a/internal/socialify/socialify.go +++ b/internal/socialify/socialify.go @@ -35,10 +35,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 +57,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 } From 8f61b9b3226fefabeb810221f6bbee99ca705f33 Mon Sep 17 00:00:00 2001 From: sigmanor Date: Mon, 17 Aug 2026 20:11:32 +0300 Subject: [PATCH 4/6] feat(api): add POST /api/message/retry Exposes the manual retry over HTTP so the dashboard can finish a partial publication. The response is 200 once the retry has run: per-integration outcomes carry the individual failures, which say more than a single status code. Input problems answer 400. --- api_docs.md | 70 ++++++++++++++++++++++++++++++++++++++++- cmd/main.go | 1 + internal/models/cron.go | 7 +++++ internal/server/api.go | 38 ++++++++++++++++++++++ 4 files changed, 115 insertions(+), 1 deletion(-) diff --git a/api_docs.md b/api_docs.md index 3867e77..a5425e2 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,68 @@ 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 the image is regenerated when the integration has `socialify_image` enabled. The posted flag is only touched if the item was still unposted. No Pushover notification is sent: a manual retry is already being watched by whoever triggered it. + +**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 — the item a partial run just consumed. | + +**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/server/api.go b/internal/server/api.go index f7d62f7..be560a1 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,42 @@ 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 + if err := json.NewDecoder(r.Body).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: From 9b3d587c618ec072d036b7019973bbfab2225ee3 Mon Sep 17 00:00:00 2001 From: sigmanor Date: Mon, 17 Aug 2026 20:48:37 +0300 Subject: [PATCH 5/6] fix(schedule): stop the message job crashing when a re-fetch is rejected The URL revalidation loop reassigned the fetch response and read it after the loop, so a rejected re-fetch left it nil and the next read panicked. The job's deferred handler re-panics, and no gocron panic handler is registered, so this took the whole process down - and skipped the posted update, leaving the item to be published a second time by the next run. Turning alchemist rejections into errors made that path routine rather than rare, so the loop now reports its own outcome instead. Run details are assembled at exit as well, so an early return or a panic still records which item was consumed and which connectors missed it; a run with nothing to retry records no details at all instead of an empty object. --- internal/repository/endpoints.go | 38 +++-- internal/repository/get_repository.go | 2 +- internal/repository/update_repository.go | 2 +- internal/repository/validate_repository.go | 4 +- internal/schedule/message-publish.go | 73 +++++++-- internal/schedule/message-publish_test.go | 97 +++++++++++ internal/schedule/message-schedule.go | 57 ++++--- internal/schedule/message-schedule_test.go | 180 +++++++++++++++++++++ internal/server/api.go | 4 +- internal/socialify/socialify.go | 6 +- internal/store/sqlite_store_test.go | 25 +++ internal/utils/remove_files.go | 12 +- internal/utils/remove_files_test.go | 42 +++++ 13 files changed, 488 insertions(+), 54 deletions(-) create mode 100644 internal/schedule/message-schedule_test.go diff --git a/internal/repository/endpoints.go b/internal/repository/endpoints.go index c23ed79..b76f801 100644 --- a/internal/repository/endpoints.go +++ b/internal/repository/endpoints.go @@ -1,18 +1,20 @@ package repository import ( + "bytes" + "context" + "io" "net/http" "os" "strconv" "strings" - "sync" "time" ) -var ( - client *http.Client - clientOnce sync.Once -) +// 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") @@ -28,12 +30,26 @@ func getContentAlchemistTimeout() time.Duration { return time.Duration(timeoutSeconds) * time.Second } -func init() { - clientOnce.Do(func() { - client = &http.Client{ - Timeout: getContentAlchemistTimeout(), - } - }) +// 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 diff --git a/internal/repository/get_repository.go b/internal/repository/get_repository.go index 179b6c0..fc8c9d2 100644 --- a/internal/repository/get_repository.go +++ b/internal/repository/get_repository.go @@ -123,7 +123,7 @@ func makeRepositoryRequest(payload getRepositoryRequest) (*repositoryResponse, e "Authorization": {authorizationHeader()}, } - resp, err := client.Do(req) + resp, err := doRequest(req) if err != nil { return nil, fmt.Errorf("error sending request: %w", err) } diff --git a/internal/repository/update_repository.go b/internal/repository/update_repository.go index 460829b..cb5b14f 100644 --- a/internal/repository/update_repository.go +++ b/internal/repository/update_repository.go @@ -35,7 +35,7 @@ func UpdateRepositoryPosted(url string, posted bool) (bool, error) { "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.go b/internal/repository/validate_repository.go index f0e14a4..ec7aeaa 100644 --- a/internal/repository/validate_repository.go +++ b/internal/repository/validate_repository.go @@ -14,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 }, @@ -52,7 +52,7 @@ func DeleteRepository(url string) (bool, error) { "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/schedule/message-publish.go b/internal/schedule/message-publish.go index 9ba9170..cfb14c0 100644 --- a/internal/schedule/message-publish.go +++ b/internal/schedule/message-publish.go @@ -12,6 +12,7 @@ import ( "os" "path/filepath" "strings" + "sync" "time" ) @@ -27,6 +28,27 @@ var retrySocialifyConfig = socialify.RetryConfig{ 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) { @@ -58,7 +80,7 @@ func publishItem(apiName string, endpoint api.APIEndpoint, item repository.Item, if endpoint.SocialifyImage && imageName != "" { publicURL := os.Getenv("PUBLIC_URL") if publicURL != "" { - req.JSONBody["image_url"] = fmt.Sprintf("%s/images/%s", publicURL, filepath.Base(imageName)) + 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) } @@ -94,8 +116,9 @@ type RetryResult struct { // 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, which is the -// item a partial run has just consumed. +// 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 { @@ -107,6 +130,9 @@ func RetryMessagePost(st store.StoreInterface, apiNames []string, url string) (* 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) @@ -122,6 +148,18 @@ func RetryMessagePost(st store.StoreInterface, apiNames []string, url string) (* 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 { @@ -145,8 +183,7 @@ func RetryMessagePost(st store.StoreInterface, apiNames []string, url string) (* } itemPosted = itemPosted || item.Posted - imageName := "" - if endpoint.SocialifyImage { + if endpoint.SocialifyImage && imageName == "" { imageName, err = generateRetryImage(item.URL) if err != nil { result.addFailure(apiName, fmt.Sprintf("failed to prepare image: %v", err)) @@ -156,14 +193,6 @@ func RetryMessagePost(st store.StoreInterface, apiNames []string, url string) (* resp, err := publishItem(apiName, endpoint, *item, imageName) - // Only this retry's own file is removed: the cron shares the directory and - // serves its image over /images/ while its own request is in flight. - if imageName != "" { - if removeErr := os.Remove(imageName); removeErr != nil && !os.IsNotExist(removeErr) { - log.Errorf("Failed to remove retry image %s: %v", imageName, removeErr) - } - } - switch { case err != nil: log.Errorf("%s API error during manual retry: %v", apiName, err) @@ -178,7 +207,10 @@ func RetryMessagePost(st store.StoreInterface, apiNames []string, url string) (* } } - if len(result.Succeeded) > 0 && !itemPosted { + // 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) } @@ -248,13 +280,24 @@ func normalizeAPINames(apiNames []string) ([]string, error) { 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("./tmp/gh_project_img/retry_%d.png", time.Now().UnixNano()) + 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) } } diff --git a/internal/schedule/message-publish_test.go b/internal/schedule/message-publish_test.go index 4595e6b..051cc0d 100644 --- a/internal/schedule/message-publish_test.go +++ b/internal/schedule/message-publish_test.go @@ -264,6 +264,103 @@ func TestRetryMessagePostRejectsEmptyAPIList(t *testing.T) { } } +// 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 { diff --git a/internal/schedule/message-schedule.go b/internal/schedule/message-schedule.go index 553431f..2feb0dd 100644 --- a/internal/schedule/message-schedule.go +++ b/internal/schedule/message-schedule.go @@ -20,22 +20,39 @@ func MessageJob(s *gocron.Scheduler, store store.StoreInterface) { var status int var logMessage string - // Recorded alongside the human-readable output so a partial run can be - // re-sent to the connectors that missed the item. - details := &models.MessageRunDetails{} + + 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.LogCronExecutionDetails("message", 0, panicMessage, details); 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.LogCronExecutionDetails("message", status, logMessage, details); 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) @@ -85,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 { @@ -102,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 @@ -134,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 { @@ -152,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 } @@ -193,10 +214,6 @@ func MessageJob(s *gocron.Scheduler, store store.StoreInterface) { } } - details.URL = updatedURL - details.Sent = successfulAPIs - details.Failed = failedAPIs - if len(successfulAPIs) > 0 && updatedURL != "" { if _, err := repository.UpdateRepositoryPosted(updatedURL, true); err != nil { log.Error("Error updating repository posted status: %v", err) @@ -206,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 be560a1..e658e82 100644 --- a/internal/server/api.go +++ b/internal/server/api.go @@ -487,7 +487,9 @@ func (api *CronAPI) RetryMessagePost(w http.ResponseWriter, r *http.Request) { } var req models.RetryMessageRequest - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + // 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 } diff --git a/internal/socialify/socialify.go b/internal/socialify/socialify.go index e068f51..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 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/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") From f8fce3d7fb7fef3a499a3b242d21c26e4743c525 Mon Sep 17 00:00:00 2001 From: sigmanor Date: Mon, 17 Aug 2026 20:49:08 +0300 Subject: [PATCH 6/6] fix(schedule): make a manual retry safe to run beside the cron Three ways the retry and the publishing cron could damage each other: - the cron's cleanup removed every file in the shared image folder, including an image a retry was still uploading, while a file the retry deleted first made that cleanup fail and report a successful run as failed. Retry images now live in a subdirectory the cleanup skips, and a file that vanished under it is no longer an error. - an image was generated per integration rather than once per retry, so each connector triggered its own upstream fetch. - the socialify client had no timeout, so bounding the attempts did not bound the request: a hung upstream blocked the synchronous endpoint indefinitely. Retries are also serialised, so a double-clicked button cannot publish twice, and the request body is size-limited. --- api_docs.md | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/api_docs.md b/api_docs.md index a5425e2..7fcea11 100644 --- a/api_docs.md +++ b/api_docs.md @@ -485,7 +485,9 @@ curl -H "Authorization: Bearer " \ 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 the image is regenerated when the integration has `socialify_image` enabled. The posted flag is only touched if the item was still unposted. No Pushover notification is sent: a manual retry is already being watched by whoever triggered it. +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:** @@ -505,7 +507,7 @@ curl -X POST \ | 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 — the item a partial run just consumed. | +| `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:**