diff --git a/api_docs.md b/api_docs.md index 7fcea11..a22aec5 100644 --- a/api_docs.md +++ b/api_docs.md @@ -487,7 +487,9 @@ A message run marks the repository as posted as soon as **any** integration succ 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. +Retries are serialised against each other, against [`/api/message/publish`](#apimessagepublish) and against the `message` cron — a double-clicked button cannot publish twice, and a cron run cannot mark an item it never sent. Unlike publish-now, a retry **waits** for whatever holds that lock instead of refusing: it is a repair action, and refusing it would leave the connectors that failed unrepaired. The wait is unbounded and there is no request timeout, so a retry issued during a cron run can hold its connection for the length of that run (minutes, in the worst case described under [`/api/message/publish`](#apimessagepublish)); repeated clicks queue up behind it rather than failing fast. + +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:** @@ -516,6 +518,8 @@ curl -X POST \ - `message`: The text recorded in cron history - `succeeded` / `failed`: Integration names per outcome - `outcomes`: Per-integration detail, with an `error` string for every failure +- `posted`: Whether the item was marked as published and left the publication queue +- `posted_error`: Present only when that marking failed — the item went out but stayed in the queue, so the scheduled run will publish it again **Response Example:** @@ -539,6 +543,75 @@ Every retry is recorded in cron history under the `message` name with `details.m - 401: Unauthorized - Invalid or missing Bearer token - 500: Internal Server Error - API configurations not loaded, or the repository could not be resolved +### /api/message/publish + +**Endpoint:** `/api/message/publish` + +**Method:** `POST` + +**Description:** Publish one repository to every enabled integration immediately, instead of waiting for its turn in the publication queue. + +It exists because a post that is lost *after* publication — a bad record deleted from the queue once it was already sent — cannot be recovered by promoting anything: the item is gone from the queue. Promoting a repository (content-alchemist's `/promote-repository/`) only moves it to the head of the queue; this endpoint publishes it now. + +There is no `apis` parameter: the run targets every integration whose `api_configs` row has `enabled = true`, resolved server-side at request time. A dashboard's cached configuration must not decide what actually gets published. + +The repository text is fetched per integration in that integration's configured `text_language`, and one image is generated for the whole run when any integration has `socialify_image` enabled. No Pushover notification is sent — whoever triggered the run is watching it. + +Unlike the message cron this does **not** revalidate the repository URL and delete dead ones: the cron picks blindly from the queue, while this publishes the row a human chose. + +**Marking as posted:** the item is marked as published as soon as **any** integration accepts it, matching the message cron (and unlike [`/api/message/retry`](#apimessageretry), which requires every requested integration to succeed). The run is recorded in cron history as a manual one, so the integrations that failed are finished off from there with the retry endpoint. + +**Concurrency:** serialised against the message cron and the retry endpoint. Rather than queueing behind a cron run that can take minutes, it refuses with `409`. A repository that is already published is refused with `409` as well, before any integration is contacted — a stale dashboard row must not publish the same post twice. + +**Duration:** the request is **synchronous** and can take minutes — image generation retries for up to ~63 s, the Threads connector alone is configured with a 90 s timeout, and every integration adds its own repository lookup. Any reverse proxy in front of content-maestro needs a matching read timeout, or the browser gets a `504` while the publication keeps running. + +**Curl Example:** + +```bash +curl -X POST \ + 'http://localhost:8080/api/message/publish' \ + -H 'Authorization: Bearer ' \ + -H 'Content-Type: application/json' \ + -d '{ + "url": "https://github.com/resemble-ai/chatterbox" + }' +``` + +**Request Parameters:** + +| Parameter | Type | Required | Description | +| --------- | ------ | -------- | ------------------------------------------------------------------------------------------------ | +| `url` | string | Yes | Repository to publish. There is no fallback: "publish something now" is never a safe guess. | + +**Response Structure:** identical to [`/api/message/retry`](#apimessageretry) — `url`, `status`, `message`, `succeeded`, `failed`, `outcomes`, `posted`, `posted_error`. + +**Response Example:** + +```json +{ + "url": "https://github.com/resemble-ai/chatterbox", + "status": 2, + "message": "Manual publish: https://github.com/resemble-ai/chatterbox sent to: telegram. Failed: threads. Errors: threads: API request failed with status 500", + "succeeded": ["telegram"], + "failed": ["threads"], + "outcomes": [ + { "api_name": "telegram", "success": true }, + { "api_name": "threads", "success": false, "error": "API request failed with status 500" } + ], + "posted": true +} +``` + +Every run is recorded in cron history under the `message` name with `details.manual = true`, so [`/api/cron-history`](#apicron-history) shows it alongside scheduled runs — and its `details.failed` list is what the retry endpoint works from. + +**Status Codes:** + +- 200: The publication ran. Individual integration failures are reported in `outcomes`, not in the status code +- 400: Bad Request - Invalid body or a missing `url` +- 401: Unauthorized - Invalid or missing Bearer token +- 409: Conflict - Another publication is already running, no integration is enabled, or the repository is already published +- 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 f0aadb4..0ad1191 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -112,6 +112,7 @@ func main() { 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/message/publish", middleware.LoggingMiddleware(middleware.CorsMiddleware(middleware.AuthMiddleware(http.HandlerFunc(cronAPI.PublishMessageNow))))) 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 0817df5..085c33d 100644 --- a/internal/models/cron.go +++ b/internal/models/cron.go @@ -28,3 +28,10 @@ type RetryMessageRequest struct { APIs []string `json:"apis"` URL string `json:"url"` } + +// PublishMessageRequest asks for one repository to be published immediately to +// every enabled integration. URL is required: there is no sensible fallback for +// "publish something now". +type PublishMessageRequest struct { + URL string `json:"url"` +} diff --git a/internal/repository/get_repository.go b/internal/repository/get_repository.go index fc8c9d2..38c2cbe 100644 --- a/internal/repository/get_repository.go +++ b/internal/repository/get_repository.go @@ -3,6 +3,7 @@ package repository import ( "bytes" "encoding/json" + "errors" "fmt" "io" "net/http" @@ -58,6 +59,12 @@ func GetRepository(limit int, posted bool, sort_order, sort_by string, textLangu }) } +// ErrRepositoryNotFound reports that content-alchemist has no repository for the +// requested identifier. It is a sentinel so HTTP callers can answer 404 for what +// is a client mistake - a typo'd or already deleted url - instead of 500, which +// would be indistinguishable from content-alchemist itself being down. +var ErrRepositoryNotFound = errors.New("repository not found") + // 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. @@ -75,7 +82,7 @@ func GetRepositoryByURL(url, textLanguage string) (*Item, error) { } if len(response.Data.Items) == 0 { - return nil, fmt.Errorf("repository %s not found", url) + return nil, fmt.Errorf("%w: %s", ErrRepositoryNotFound, url) } item := response.Data.Items[0] @@ -137,6 +144,12 @@ func makeRepositoryRequest(payload getRepositoryRequest) (*repositoryResponse, e // 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.StatusNotFound { + // content-alchemist answers 404 for an identifier it does not know, which is + // a client mistake rather than a failure of either service. + return nil, fmt.Errorf("%w: %s", ErrRepositoryNotFound, errorDetail(respBody)) + } + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { return nil, fmt.Errorf("content-alchemist error (status %d): %s", resp.StatusCode, errorDetail(respBody)) } diff --git a/internal/schedule/message-publish.go b/internal/schedule/message-publish.go index cfb14c0..7a6d83b 100644 --- a/internal/schedule/message-publish.go +++ b/internal/schedule/message-publish.go @@ -11,14 +11,31 @@ import ( "fmt" "os" "path/filepath" + "sort" "strings" "sync" "time" ) -// ErrInvalidRetryRequest marks a retry that was rejected because of its input, -// so callers can answer with 400 instead of 500. -var ErrInvalidRetryRequest = errors.New("invalid retry request") +// ErrInvalidRetryRequest marks a manual publication that was rejected because of +// its input, so callers can answer with 400 instead of 500. Its message says +// "request" rather than "retry": the dashboard shows it verbatim, and both manual +// endpoints return it. +var ErrInvalidRetryRequest = errors.New("invalid request") + +// ErrPublishBusy reports that another publication holds publishMutex. Publishing +// on demand answers an HTTP request, so it refuses rather than queueing behind a +// cron run that can take minutes. +var ErrPublishBusy = errors.New("another publication is already running") + +// ErrNoEnabledIntegrations reports that no integration is enabled, so a run that +// targets all of them has nothing to send to. +var ErrNoEnabledIntegrations = errors.New("no integration is enabled") + +// ErrAlreadyPosted reports that the item left the publication queue between the +// dashboard rendering the row and the request arriving - a cron run in between, +// or a second dashboard. Publishing it again would duplicate the post. +var ErrAlreadyPosted = errors.New("repository is already published") // retrySocialifyConfig keeps image generation short: a manual retry answers an // HTTP request, and the cron default (5 attempts, 20s apart) would block it for @@ -33,10 +50,11 @@ const ( 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 +// publishMutex serialises everything that publishes an item: the message cron and +// both manual endpoints. Two of them publishing the same item concurrently - a +// double-clicked button, two open dashboards, or a manual run landing mid-cron - +// would post twice to the same connector. +var publishMutex 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. @@ -95,21 +113,57 @@ func publishItem(apiName string, endpoint api.APIEndpoint, item repository.Item, return api.ExecuteRequest(req) } -// RetryOutcome is the per-connector result of a manual retry. -type RetryOutcome struct { +// PublishOutcome is the per-connector result of a manual publication. +type PublishOutcome 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"` +// PublishResult is the response of a manual publication. +type PublishResult struct { + URL string `json:"url"` + Status int `json:"status"` + Message string `json:"message"` + Succeeded []string `json:"succeeded"` + Failed []string `json:"failed"` + Outcomes []PublishOutcome `json:"outcomes"` + // Posted reports that the item was marked as published and left the queue. + Posted bool `json:"posted"` + // PostedError explains why that marking failed. It is the one failure the + // per-connector outcomes cannot show: the item went out but stayed in the + // queue, so the scheduled run will publish it again. + PostedError string `json:"posted_error,omitempty"` +} + +// markPostedPolicy decides when a still-unposted item leaves the publication +// queue after a manual run. +type markPostedPolicy int + +const ( + // markPostedWhenComplete is the retry policy: the item stays in the queue + // until every requested connector has it. A retry exists to repair a partial + // run, so marking it posted after another partial run re-creates the failure. + markPostedWhenComplete markPostedPolicy = iota + // markPostedOnAnySuccess is the cron policy: one connector receiving the item + // is a publication. The connectors that failed are finished off from cron + // history with the retry endpoint, which is why the run is recorded as manual. + markPostedOnAnySuccess +) + +// publishOptions describes one manual publication run. +type publishOptions struct { + // url pins the repository before any connector is contacted. Empty falls back + // to the most recently published one. + url string + // apiNames selects the connectors. Empty means every enabled one. + apiNames []string + // requireUnposted refuses an item that already left the queue, so a stale row + // in one dashboard cannot re-publish what another one - or the cron - just sent. + requireUnposted bool + markPosted markPostedPolicy + // label prefixes the cron-history message: "Manual retry" / "Manual publish". + label string } // RetryMessagePost re-sends one repository to the named APIs. It exists because a @@ -119,23 +173,81 @@ type RetryResult struct { // 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) { +func RetryMessagePost(st store.StoreInterface, apiNames []string, url string) (*PublishResult, error) { + requested, err := normalizeAPINames(apiNames) + if err != nil { + return nil, err + } + + // A retry waits for whatever is publishing right now: it is a repair action, and + // refusing it would leave the connectors that failed unrepaired. + publishMutex.Lock() + defer publishMutex.Unlock() + + return publishManually(st, publishOptions{ + url: url, + apiNames: requested, + markPosted: markPostedWhenComplete, + label: "Manual retry", + }) +} + +// PublishNow publishes one repository to every enabled integration immediately, +// instead of waiting for its turn in the publication queue. It exists because a +// post that was lost after publication - a bad record deleted from the queue - +// cannot be recovered by promoting anything: the item is already gone from it. +// +// The item is marked as posted as soon as any integration accepts it, matching the +// message cron. The run is recorded as manual, so the connectors that failed are +// finished off from cron history with RetryMessagePost. +// +// Unlike the cron this does not revalidate the repository URL and delete dead +// ones: the cron picks blindly from the queue, while this publishes the row a +// human chose, and deleting that row mid-request would be astonishing. +func PublishNow(st store.StoreInterface, url string) (*PublishResult, error) { + url = strings.TrimSpace(url) + if url == "" { + return nil, fmt.Errorf("%w: url is required", ErrInvalidRetryRequest) + } + + // Unlike a retry this refuses to queue: it answers a request with a spinner in + // front of it, and a cron run ahead of it can hold the lock for minutes. + if !publishMutex.TryLock() { + return nil, ErrPublishBusy + } + defer publishMutex.Unlock() + + return publishManually(st, publishOptions{ + url: url, + requireUnposted: true, + markPosted: markPostedOnAnySuccess, + label: "Manual publish", + }) +} + +// publishManually publishes one repository to a set of connectors and records the +// run in cron history under the message job. Shared by the retry and publish-now +// endpoints, which differ only in how they choose connectors and in when a +// still-unposted item is allowed to leave the queue. +// +// The caller must already hold publishMutex. +func publishManually(st store.StoreInterface, opts publishOptions) (*PublishResult, 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 + requested := opts.apiNames + if len(requested) == 0 { + requested = enabledAPINames(apiConfigs) + if len(requested) == 0 { + return nil, ErrNoEnabledIntegrations + } } - 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) + url := strings.TrimSpace(opts.url) itemPosted := false if url == "" { latest, err := repository.GetLatestPostedRepository("") @@ -146,9 +258,25 @@ func RetryMessagePost(st store.StoreInterface, apiNames []string, url string) (* itemPosted = latest.Posted } - result := &RetryResult{URL: url} + if opts.requireUnposted { + // One lookup before anything is published, deliberately kept even though the + // connector loop fetches the item again a moment later: it is what makes an + // item that already left the queue, and a url content-alchemist does not + // know, a single clear error instead of a set of per-connector failures plus + // a cron-history row for a run that could never have worked. The language is + // empty because only the posted flag is read here. + current, err := repository.GetRepositoryByURL(url, "") + if err != nil { + return nil, fmt.Errorf("failed to resolve %s: %w", url, err) + } + if current.Posted { + return nil, fmt.Errorf("%w: %s", ErrAlreadyPosted, url) + } + } - // One image per retry, not one per connector: the cron shares a single image + result := &PublishResult{URL: url} + + // One image per run, 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() { @@ -156,7 +284,7 @@ func RetryMessagePost(st store.StoreInterface, apiNames []string, url string) (* return } if err := os.Remove(imageName); err != nil && !os.IsNotExist(err) { - log.Errorf("Failed to remove retry image %s: %v", imageName, err) + log.Errorf("Failed to remove manual publication image %s: %v", imageName, err) } }() @@ -195,42 +323,45 @@ func RetryMessagePost(st store.StoreInterface, apiNames []string, url string) (* switch { case err != nil: - log.Errorf("%s API error during manual retry: %v", apiName, err) + log.Errorf("%s API error during %s: %v", apiName, opts.label, 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}) + log.Debugf("%s post created successfully during %s with language %s!", apiName, opts.label, textLanguage) + result.addSuccess(apiName) default: - log.Errorf("%s API request failed during manual retry (status %d): %s", apiName, resp.StatusCode, string(resp.Body)) + log.Errorf("%s API request failed during %s (status %d): %s", apiName, opts.label, resp.StatusCode, string(resp.Body)) result.addFailure(apiName, fmt.Sprintf("API request failed with status %d", resp.StatusCode)) } } - // Marking an unposted item as posted while some connector still failed would - // drop it out of the queue - the very failure this endpoint exists to repair - - // so it is only marked once every requested connector has it. - if !itemPosted && len(result.Succeeded) > 0 && len(result.Failed) == 0 { - if _, err := repository.UpdateRepositoryPosted(url, true); err != nil { - log.Errorf("Failed to update posted status for %s after manual retry: %v", url, err) + if !itemPosted && shouldMarkPosted(opts.markPosted, result) { + if ok, err := repository.UpdateRepositoryPosted(url, true); err != nil || !ok { + // The item published but stayed in the queue, so the scheduled run will + // publish it again. Report it: this is the one failure the per-connector + // outcomes cannot show. + result.PostedError = postedErrorMessage(err) + log.Errorf("Failed to update posted status for %s after %s: %v", url, opts.label, err) + } else { + result.Posted = true } } switch { case len(result.Succeeded) == 0: result.Status = 0 - result.Message = fmt.Sprintf("Manual retry: nothing sent for %s. Errors: %s", url, result.errorSummary()) + result.Message = fmt.Sprintf("%s: nothing sent for %s. Errors: %s", opts.label, 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()) + result.Message = fmt.Sprintf("%s: %s sent to: %s. Failed: %s. Errors: %s", + opts.label, 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, ", ")) + result.Message = fmt.Sprintf("%s: %s sent to: %s", opts.label, 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 + // Recorded under the message job so the dashboard's existing filters show it, + // and so the failed connectors can be finished off with the retry endpoint. + // No Pushover notification: a manual run is already being watched by whoever // triggered it. details := &models.MessageRunDetails{ URL: url, @@ -239,18 +370,23 @@ func RetryMessagePost(st store.StoreInterface, apiNames []string, url string) (* Manual: true, } if err := st.LogCronExecutionDetails("message", result.Status, result.Message, details); err != nil { - log.Errorf("Failed to log manual retry execution: %v", err) + log.Errorf("Failed to log %s execution: %v", opts.label, err) } return result, nil } -func (r *RetryResult) addFailure(apiName, message string) { +func (r *PublishResult) addSuccess(apiName string) { + r.Succeeded = append(r.Succeeded, apiName) + r.Outcomes = append(r.Outcomes, PublishOutcome{APIName: apiName, Success: true}) +} + +func (r *PublishResult) addFailure(apiName, message string) { r.Failed = append(r.Failed, apiName) - r.Outcomes = append(r.Outcomes, RetryOutcome{APIName: apiName, Success: false, Error: message}) + r.Outcomes = append(r.Outcomes, PublishOutcome{APIName: apiName, Success: false, Error: message}) } -func (r *RetryResult) errorSummary() string { +func (r *PublishResult) errorSummary() string { messages := make([]string, 0, len(r.Outcomes)) for _, outcome := range r.Outcomes { if !outcome.Success { @@ -260,6 +396,39 @@ func (r *RetryResult) errorSummary() string { return strings.Join(messages, "; ") } +// shouldMarkPosted applies the caller's policy to a finished run. +func shouldMarkPosted(policy markPostedPolicy, result *PublishResult) bool { + if len(result.Succeeded) == 0 { + return false + } + return policy == markPostedOnAnySuccess || len(result.Failed) == 0 +} + +// postedErrorMessage describes a failed posted-status update. content-alchemist +// answering "not ok" without an error of its own still means the item stayed in +// the queue, so it needs wording too. +func postedErrorMessage(err error) string { + if err != nil { + return err.Error() + } + return "content-alchemist did not confirm the posted status" +} + +// enabledAPINames lists the enabled connectors in a stable order. The config is a +// map, so without sorting the connectors would be contacted - and reported - in a +// different order on every call. +func enabledAPINames(configs *api.APIConfig) []string { + names := make([]string, 0, len(configs.APIs)) + for name, endpoint := range configs.APIs { + if endpoint.Enabled { + names = append(names, name) + } + } + sort.Strings(names) + + return names +} + func normalizeAPINames(apiNames []string) ([]string, error) { seen := make(map[string]bool, len(apiNames)) normalized := make([]string, 0, len(apiNames)) diff --git a/internal/schedule/message-publish_test.go b/internal/schedule/message-publish_test.go index 051cc0d..c7cccf3 100644 --- a/internal/schedule/message-publish_test.go +++ b/internal/schedule/message-publish_test.go @@ -3,9 +3,11 @@ package schedule import ( "content-maestro/internal/api" "content-maestro/internal/models" + "content-maestro/internal/repository" "content-maestro/internal/store" "encoding/json" "errors" + "fmt" "net/http" "net/http/httptest" "strings" @@ -377,3 +379,386 @@ func TestNormalizeAPINames(t *testing.T) { } } } + +// publishAlchemistStub answers get-repository for a single known repository and +// counts the update-posted calls, which is what pins the mark-posted policy. +func publishAlchemistStub(t *testing.T, posted bool, languages *[]string, patches *int) *httptest.Server { + t.Helper() + + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodPatch { + if patches != nil { + *patches++ + } + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"status":"ok","message":"updated"}`)) + return + } + + var body struct { + URL string `json:"url"` + 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 languages != nil { + *languages = append(*languages, body.TextLanguage) + } + + 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": posted, + "url": url, + "text": "text for " + body.TextLanguage, + }}, + }, + }) + })) +} + +func TestPublishNowSendsToEveryEnabledIntegration(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 + patches := 0 + alchemist := publishAlchemistStub(t, false, &languages, &patches) + 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: "telegram", URL: connector.URL, Method: http.MethodPost, + ContentType: "json", SuccessCode: http.StatusOK, Enabled: true, + TextLanguage: "uk", + }, + { + Name: "bluesky", URL: connector.URL, 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 := PublishNow(st, retryTestURL) + if err != nil { + t.Fatalf("PublishNow() error = %v", err) + } + + if len(connectorBodies) != 2 { + t.Fatalf("connector received %d requests, want 2 (one per enabled integration)", len(connectorBodies)) + } + + texts := map[string]bool{} + for _, body := range connectorBodies { + if got := body["url"]; got != retryTestURL { + t.Errorf("published url = %v, want %v", got, retryTestURL) + } + texts[fmt.Sprint(body["text"])] = true + } + if !texts["text for en"] || !texts["text for uk"] { + t.Errorf("published texts = %v, want both the English and the Ukrainian text", texts) + } + + if result.Status != 1 { + t.Errorf("status = %d, want 1", result.Status) + } + for _, outcome := range result.Outcomes { + if outcome.APIName == "bluesky" { + t.Errorf("outcomes = %+v, want no entry for the disabled integration", result.Outcomes) + } + } + // Ordering is what lets the dashboard list the integrations the same way the + // run reports them, so it is asserted rather than merely relied upon. + if len(result.Succeeded) != 2 || result.Succeeded[0] != "telegram" || result.Succeeded[1] != "threads" { + t.Errorf("succeeded = %v, want [telegram threads]", result.Succeeded) + } + if !result.Posted || patches != 1 { + t.Errorf("posted = %v after %d update-posted calls, want it marked once", result.Posted, patches) + } + + 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 publish:") { + t.Errorf("history output = %q, want a Manual publish 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) + } +} + +// Publishing on demand follows the cron: one integration accepting the item is a +// publication, so the item leaves the queue and the connectors that failed are +// finished off from cron history. Without this the next scheduled run would +// publish it again to the integrations that already have it. +func TestPublishNowMarksPostedOnPartialSuccess(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() + + patches := 0 + alchemist := publishAlchemistStub(t, false, nil, &patches) + 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: "twitter", URL: "http://127.0.0.1:1", 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) + } + + result, err := PublishNow(st, retryTestURL) + if err != nil { + t.Fatalf("PublishNow() error = %v", err) + } + + if result.Status != 2 { + t.Errorf("status = %d, want 2 (partial)", result.Status) + } + if len(result.Succeeded) != 1 || result.Succeeded[0] != "threads" { + t.Errorf("succeeded = %v, want [threads]", result.Succeeded) + } + if len(result.Failed) != 1 || result.Failed[0] != "twitter" { + t.Errorf("failed = %v, want [twitter]", result.Failed) + } + if patches != 1 || !result.Posted { + t.Errorf("update-posted calls = %d, posted = %v; want it marked once", patches, result.Posted) + } + if result.PostedError != "" { + t.Errorf("posted error = %q, want none", result.PostedError) + } + // The failed integration has to reach cron history, or the retry button in the + // dashboard has nothing to finish off. + if st.loggedDetails == nil || len(st.loggedDetails.Failed) != 1 || st.loggedDetails.Failed[0] != "twitter" { + t.Errorf("history details = %+v, want twitter recorded as failed", st.loggedDetails) + } +} + +func TestPublishNowDoesNotMarkPostedWhenNothingSent(t *testing.T) { + patches := 0 + alchemist := publishAlchemistStub(t, false, nil, &patches) + defer alchemist.Close() + withRepositoryEndpoints(t, alchemist.URL) + + st := &retryStore{configs: []models.APIConfigModel{{ + Name: "threads", URL: "http://127.0.0.1:1", 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) + } + + result, err := PublishNow(st, retryTestURL) + if err != nil { + t.Fatalf("PublishNow() error = %v", err) + } + + if result.Status != 0 { + t.Errorf("status = %d, want 0", result.Status) + } + if patches != 0 || result.Posted { + t.Errorf("update-posted calls = %d, posted = %v; want the item left in the queue", patches, result.Posted) + } + if st.logCalls != 1 { + t.Errorf("cron history writes = %d, want 1", st.logCalls) + } +} + +func TestPublishNowRejectsMissingURL(t *testing.T) { + for _, url := range []string{"", " "} { + st := &retryStore{} + if err := api.LoadAPIConfigs(st); err != nil { + t.Fatalf("LoadAPIConfigs() error = %v", err) + } + + _, err := PublishNow(st, url) + if !errors.Is(err, ErrInvalidRetryRequest) { + t.Fatalf("PublishNow(%q) error = %v, want ErrInvalidRetryRequest", url, err) + } + if st.logCalls != 0 { + t.Errorf("cron history writes = %d, want none for a rejected request", st.logCalls) + } + } +} + +func TestPublishNowRejectsWhenNoIntegrationEnabled(t *testing.T) { + var connectorRequests int + connector := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + connectorRequests++ + w.WriteHeader(http.StatusOK) + })) + defer connector.Close() + + st := &retryStore{configs: []models.APIConfigModel{{ + Name: "threads", URL: connector.URL, 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) + } + + _, err := PublishNow(st, retryTestURL) + if !errors.Is(err, ErrNoEnabledIntegrations) { + t.Fatalf("PublishNow() error = %v, want ErrNoEnabledIntegrations", err) + } + if connectorRequests != 0 { + t.Errorf("connector requests = %d, want none", connectorRequests) + } + if st.logCalls != 0 { + t.Errorf("cron history writes = %d, want none", st.logCalls) + } +} + +// An item that already left the queue must not be published again: the row can be +// stale in a second dashboard, or a cron run can have consumed it seconds earlier. +func TestPublishNowRefusesAlreadyPostedRepository(t *testing.T) { + var connectorRequests int + connector := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + connectorRequests++ + w.WriteHeader(http.StatusOK) + })) + defer connector.Close() + + alchemist := publishAlchemistStub(t, true, nil, nil) + 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) + } + + _, err := PublishNow(st, retryTestURL) + if !errors.Is(err, ErrAlreadyPosted) { + t.Fatalf("PublishNow() error = %v, want ErrAlreadyPosted", err) + } + if connectorRequests != 0 { + t.Errorf("connector requests = %d, want none - nothing may be published twice", connectorRequests) + } + if st.logCalls != 0 { + t.Errorf("cron history writes = %d, want none", st.logCalls) + } +} + +// A url content-alchemist does not know is a client mistake, so it has to be +// distinguishable from content-alchemist being unreachable - the handler answers +// 404 for the first and 500 for the second. +func TestPublishNowReportsAnUnknownRepositoryAsNotFound(t *testing.T) { + var connectorRequests int + connector := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + connectorRequests++ + w.WriteHeader(http.StatusOK) + })) + defer connector.Close() + + alchemist := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusNotFound) + w.Write([]byte(`{"status":"error","message":"Repository not found"}`)) + })) + 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) + } + + _, err := PublishNow(st, "https://github.com/nobody/nothing") + if !errors.Is(err, repository.ErrRepositoryNotFound) { + t.Fatalf("PublishNow() error = %v, want repository.ErrRepositoryNotFound", err) + } + if connectorRequests != 0 { + t.Errorf("connector requests = %d, want none", connectorRequests) + } + if st.logCalls != 0 { + t.Errorf("cron history writes = %d, want none for a run that never started", st.logCalls) + } +} + +// Publishing on demand refuses instead of queueing: a cron run ahead of it can +// hold the lock for minutes, and there is a spinner in front of the request. +func TestPublishNowRefusesConcurrentRun(t *testing.T) { + publishMutex.Lock() + defer publishMutex.Unlock() + + st := &retryStore{} + if _, err := PublishNow(st, retryTestURL); !errors.Is(err, ErrPublishBusy) { + t.Fatalf("PublishNow() error = %v, want ErrPublishBusy", err) + } + if st.logCalls != 0 { + t.Errorf("cron history writes = %d, want none", st.logCalls) + } +} + +func TestEnabledAPINames(t *testing.T) { + configs := &api.APIConfig{APIs: map[string]api.APIEndpoint{ + "threads": {Enabled: true}, + "bluesky": {Enabled: true}, + "telegram": {Enabled: false}, + }} + + got := enabledAPINames(configs) + want := []string{"bluesky", "threads"} + if len(got) != len(want) { + t.Fatalf("enabledAPINames() = %v, want %v", got, want) + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("enabledAPINames() = %v, want %v", got, want) + } + } +} diff --git a/internal/schedule/message-schedule.go b/internal/schedule/message-schedule.go index 2feb0dd..2e4aae7 100644 --- a/internal/schedule/message-schedule.go +++ b/internal/schedule/message-schedule.go @@ -18,6 +18,12 @@ import ( func MessageJob(s *gocron.Scheduler, store store.StoreInterface) { log.Debug("cron job started") + // Serialised against the manual publication endpoints: a manual publish that + // marks its item posted mid-run would leave this run marking an item it never + // sent. Registered before the recover defer below, so a panic still unlocks. + publishMutex.Lock() + defer publishMutex.Unlock() + var status int var logMessage string diff --git a/internal/server/api.go b/internal/server/api.go index e658e82..5d90118 100644 --- a/internal/server/api.go +++ b/internal/server/api.go @@ -3,6 +3,7 @@ package server import ( apiExecutor "content-maestro/internal/api" "content-maestro/internal/models" + "content-maestro/internal/repository" "content-maestro/internal/schedule" "content-maestro/internal/store" "content-maestro/internal/validation" @@ -508,6 +509,61 @@ func (api *CronAPI) RetryMessagePost(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(result) } +// PublishMessageNow publishes a repository immediately instead of waiting for its +// turn in the publication queue. It exists because a post lost after publication - +// a bad record deleted from the queue - cannot be recovered by promoting anything. +// +// It targets every enabled integration and takes no API list: the dashboard's +// cached configuration must not decide what actually gets published. +// +// Like the retry endpoint this answers 200 once the run happened, however many +// integrations failed - the per-integration outcomes are the useful signal. The +// error statuses all describe why nothing was attempted at all: a rejected +// request (400), a url content-alchemist does not know (404), or a state that +// makes publishing this item now impossible (409). +func (api *CronAPI) PublishMessageNow(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.PublishMessageRequest + // The body is a single url; anything larger is not a request this endpoint + // should read into memory. + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 8*1024)).Decode(&req); err != nil { + http.Error(w, "Invalid request body", http.StatusBadRequest) + return + } + + result, err := schedule.PublishNow(api.store, req.URL) + if err != nil { + switch { + case errors.Is(err, schedule.ErrInvalidRetryRequest): + http.Error(w, err.Error(), http.StatusBadRequest) + case errors.Is(err, repository.ErrRepositoryNotFound): + // A url content-alchemist does not know is a client mistake - a typo, or a + // row deleted since the dashboard rendered it - not a server failure. + http.Error(w, err.Error(), http.StatusNotFound) + case errors.Is(err, schedule.ErrPublishBusy), + errors.Is(err, schedule.ErrNoEnabledIntegrations), + errors.Is(err, schedule.ErrAlreadyPosted): + // Nothing about the request is wrong; the server is in a state that makes + // publishing this item now impossible, and the message says which. + http.Error(w, err.Error(), http.StatusConflict) + default: + http.Error(w, err.Error(), http.StatusInternalServerError) + } + return + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(result) +} + func (api *CronAPI) HandleAPIConfigs(w http.ResponseWriter, r *http.Request) { switch r.Method { case http.MethodOptions: diff --git a/internal/server/api_test.go b/internal/server/api_test.go new file mode 100644 index 0000000..ccb9d89 --- /dev/null +++ b/internal/server/api_test.go @@ -0,0 +1,271 @@ +package server + +import ( + apiExecutor "content-maestro/internal/api" + "content-maestro/internal/models" + "content-maestro/internal/schedule" + "content-maestro/internal/store" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +// publishStore is a StoreInterface stub that only serves API configs and records +// what a run wrote to the cron history. Every other method is a hard error, so a +// handler reaching for one fails loudly instead of silently reading a zero value. +type publishStore struct { + configs []models.APIConfigModel + logCalls int +} + +func (s *publishStore) GetAllAPIConfigs() ([]models.APIConfigModel, error) { + return s.configs, nil +} + +func (s *publishStore) LogCronExecutionDetails(string, int, string, *models.MessageRunDetails) error { + s.logCalls++ + return nil +} + +func (s *publishStore) LogCronExecution(name string, status int, output string) error { + return s.LogCronExecutionDetails(name, status, output, nil) +} + +func (s *publishStore) Close() error { return nil } +func (s *publishStore) InitializeDefaultSettings() error { return nil } +func (s *publishStore) GetCronSetting(string) (*models.CronSetting, error) { + return nil, errors.New("not implemented") +} +func (s *publishStore) GetAllCronSettings() ([]models.CronSetting, error) { + return nil, errors.New("not implemented") +} +func (s *publishStore) UpdateCronSetting(string, string, bool) (*models.CronSetting, error) { + return nil, errors.New("not implemented") +} +func (s *publishStore) GetCronHistoryCount(string, *int, *time.Time, *time.Time) (int, error) { + return 0, errors.New("not implemented") +} +func (s *publishStore) GetCronHistory(string, *int, int, int, string, *time.Time, *time.Time) ([]models.CronHistory, error) { + return nil, errors.New("not implemented") +} +func (s *publishStore) GetCollectSettings() (*store.CollectSettings, error) { + return nil, errors.New("not implemented") +} +func (s *publishStore) UpdateCollectSettings(*store.CollectSettings) error { + return errors.New("not implemented") +} +func (s *publishStore) GetPromptSettings() (*models.PromptSettings, error) { + return nil, errors.New("not implemented") +} +func (s *publishStore) UpdatePromptSettings(*models.UpdatePromptSettingsRequest) error { + return errors.New("not implemented") +} +func (s *publishStore) GetAPIConfig(string) (*models.APIConfigModel, error) { + return nil, errors.New("not implemented") +} +func (s *publishStore) CreateAPIConfig(*models.CreateAPIConfigRequest) (*models.APIConfigModel, error) { + return nil, errors.New("not implemented") +} +func (s *publishStore) UpdateAPIConfig(string, *models.UpdateAPIConfigRequest) (*models.APIConfigModel, error) { + return nil, errors.New("not implemented") +} +func (s *publishStore) DeleteAPIConfig(string) error { return errors.New("not implemented") } + +var _ store.StoreInterface = (*publishStore)(nil) + +const publishTestURL = "https://github.com/resemble-ai/chatterbox" + +// The handler is called directly rather than through the middleware chain, so +// these tests cover the request validation and the error-to-status mapping only. +func newPublishAPI(t *testing.T, configs []models.APIConfigModel) (*CronAPI, *publishStore) { + t.Helper() + + st := &publishStore{configs: configs} + if err := apiExecutor.LoadAPIConfigs(st); err != nil { + t.Fatalf("LoadAPIConfigs() error = %v", err) + } + + return NewCronAPI(st, nil, nil), st +} + +func publishRequest(method, body string) *http.Request { + return httptest.NewRequest(method, "/api/message/publish", strings.NewReader(body)) +} + +func TestPublishMessageNowRejectsBadRequests(t *testing.T) { + tests := []struct { + name string + method string + body string + wantStatus int + }{ + {name: "wrong method", method: http.MethodGet, body: "", wantStatus: http.StatusMethodNotAllowed}, + {name: "malformed body", method: http.MethodPost, body: "{", wantStatus: http.StatusBadRequest}, + {name: "missing url", method: http.MethodPost, body: `{}`, wantStatus: http.StatusBadRequest}, + {name: "blank url", method: http.MethodPost, body: `{"url":" "}`, wantStatus: http.StatusBadRequest}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + api, st := newPublishAPI(t, nil) + + recorder := httptest.NewRecorder() + api.PublishMessageNow(recorder, publishRequest(tt.method, tt.body)) + + if recorder.Code != tt.wantStatus { + t.Errorf("status = %d, want %d (body %q)", recorder.Code, tt.wantStatus, recorder.Body.String()) + } + if st.logCalls != 0 { + t.Errorf("cron history writes = %d, want none for a rejected request", st.logCalls) + } + }) + } +} + +// Nothing about the request is wrong when no integration is enabled - the server +// simply cannot honour it - so it answers 409 rather than 400. +func TestPublishMessageNowConflictsWhenNoIntegrationEnabled(t *testing.T) { + api, _ := newPublishAPI(t, []models.APIConfigModel{{ + Name: "threads", URL: "http://127.0.0.1:1", Method: http.MethodPost, + ContentType: "json", SuccessCode: http.StatusOK, Enabled: false, + }}) + + recorder := httptest.NewRecorder() + api.PublishMessageNow(recorder, publishRequest(http.MethodPost, `{"url":"`+publishTestURL+`"}`)) + + if recorder.Code != http.StatusConflict { + t.Fatalf("status = %d, want %d (body %q)", recorder.Code, http.StatusConflict, recorder.Body.String()) + } + if !strings.Contains(recorder.Body.String(), schedule.ErrNoEnabledIntegrations.Error()) { + t.Errorf("body = %q, want it to name the missing integrations", recorder.Body.String()) + } +} + +// A url content-alchemist does not know is a client mistake; answering 500 would +// make it indistinguishable from content-alchemist being down. +func TestPublishMessageNowAnswersNotFoundForAnUnknownRepository(t *testing.T) { + alchemist := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusNotFound) + w.Write([]byte(`{"status":"error","message":"Repository not found"}`)) + })) + defer alchemist.Close() + t.Setenv("CONTENT_ALCHEMIST_URL", alchemist.URL) + t.Setenv("CONTENT_ALCHEMIST_BEARER", "test-token") + + api, st := newPublishAPI(t, []models.APIConfigModel{{ + Name: "threads", URL: "http://127.0.0.1:1", Method: http.MethodPost, + ContentType: "json", SuccessCode: http.StatusOK, Enabled: true, + TextLanguage: "en", + }}) + + recorder := httptest.NewRecorder() + api.PublishMessageNow(recorder, publishRequest(http.MethodPost, `{"url":"https://github.com/nobody/nothing"}`)) + + if recorder.Code != http.StatusNotFound { + t.Fatalf("status = %d, want %d (body %q)", recorder.Code, http.StatusNotFound, recorder.Body.String()) + } + if st.logCalls != 0 { + t.Errorf("cron history writes = %d, want none", st.logCalls) + } +} + +// content-alchemist being unreachable is a server failure, and must not be +// reported as the client's mistake. +func TestPublishMessageNowAnswersServerErrorWhenAlchemistIsDown(t *testing.T) { + t.Setenv("CONTENT_ALCHEMIST_URL", "http://127.0.0.1:1") + t.Setenv("CONTENT_ALCHEMIST_BEARER", "test-token") + + api, _ := newPublishAPI(t, []models.APIConfigModel{{ + Name: "threads", URL: "http://127.0.0.1:1", Method: http.MethodPost, + ContentType: "json", SuccessCode: http.StatusOK, Enabled: true, + TextLanguage: "en", + }}) + + recorder := httptest.NewRecorder() + api.PublishMessageNow(recorder, publishRequest(http.MethodPost, `{"url":"`+publishTestURL+`"}`)) + + if recorder.Code != http.StatusInternalServerError { + t.Fatalf("status = %d, want %d (body %q)", recorder.Code, http.StatusInternalServerError, recorder.Body.String()) + } +} + +// A partially failed run is still a run that happened: the per-integration +// outcomes carry the failures, so the response stays 200. +func TestPublishMessageNowAnswersOKWithOutcomes(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() + + alchemist := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodPatch { + 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) + + 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": body.URL, "text": "text", + }}, + }, + }) + })) + defer alchemist.Close() + t.Setenv("CONTENT_ALCHEMIST_URL", alchemist.URL) + t.Setenv("CONTENT_ALCHEMIST_BEARER", "test-token") + + api, st := newPublishAPI(t, []models.APIConfigModel{ + { + Name: "threads", URL: connector.URL, Method: http.MethodPost, + ContentType: "json", SuccessCode: http.StatusOK, Enabled: true, + TextLanguage: "en", + }, + { + Name: "twitter", URL: "http://127.0.0.1:1", Method: http.MethodPost, + ContentType: "json", SuccessCode: http.StatusOK, Enabled: true, + TextLanguage: "en", + }, + }) + + recorder := httptest.NewRecorder() + api.PublishMessageNow(recorder, publishRequest(http.MethodPost, `{"url":"`+publishTestURL+`"}`)) + + if recorder.Code != http.StatusOK { + t.Fatalf("status = %d, want 200 (body %q)", recorder.Code, recorder.Body.String()) + } + if got := recorder.Header().Get("Content-Type"); got != "application/json" { + t.Errorf("content type = %q, want application/json", got) + } + + var result schedule.PublishResult + if err := json.NewDecoder(recorder.Body).Decode(&result); err != nil { + t.Fatalf("failed to decode the response: %v", err) + } + if result.Status != 2 { + t.Errorf("status = %d, want 2 (partial)", result.Status) + } + if len(result.Outcomes) != 2 { + t.Errorf("outcomes = %+v, want one per integration", result.Outcomes) + } + if !result.Posted { + t.Errorf("posted = false, want the item marked as published") + } + if st.logCalls != 1 { + t.Errorf("cron history writes = %d, want 1", st.logCalls) + } +}