feat(reposts): add durable multi-stage repost cycles

backend/internal/database/migrations/132_multi_stage_reposts.sql

@@ -0,0 +1,6 @@
+ALTER TABLE repost_policies ADD COLUMN stages_json TEXT NOT NULL DEFAULT '[]';
+
+ALTER TABLE repost_executions ADD COLUMN current_stage INTEGER NOT NULL DEFAULT 1;
+ALTER TABLE repost_executions ADD COLUMN total_stages INTEGER NOT NULL DEFAULT 1;
+ALTER TABLE repost_executions ADD COLUMN unrepost_attempts INTEGER NOT NULL DEFAULT 0;
+ALTER TABLE repost_executions ADD COLUMN stage_history_json TEXT NOT NULL DEFAULT '[]';

backend/internal/database/migrations/migration_chain_test.go

@@ -9,6 +9,7 @@ import (
 	"sort"
 	"strings"
 	"testing"
+	"testing/fstest"
 	"time"
 
 	_ "github.com/mattn/go-sqlite3"
@@ -143,6 +144,111 @@ func TestMigrationChainAppliesCleanlyOnPostgres(t *testing.T) {
 	require.Positive(t, applied, "the chain must record applied migrations")
 }
 
+func TestMultiStageRepostMigrationPreservesLegacyExecutions(t *testing.T) {
+	t.Parallel()
+
+	db := newMigrationsTestDB(t)
+	ctx := t.Context()
+	for _, model := range []any{
+		(*models.Job)(nil),
+		(*models.AnalyticsAccountSnapshot)(nil),
+		(*models.AnalyticsRenditionSnapshot)(nil),
+		(*models.AnalyticsSyncState)(nil),
+	} {
+		_, err := db.NewCreateTable().Model(model).IfNotExists().Exec(ctx)
+		require.NoError(t, err)
+	}
+
+	beforeMultiStage := fstest.MapFS{}
+	entries, err := migrationFiles.ReadDir(".")
+	require.NoError(t, err)
+	for _, entry := range entries {
+		if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".sql") {
+			continue
+		}
+		version, parseErr := parseVersion(entry.Name())
+		require.NoError(t, parseErr)
+		if version >= 132 {
+			continue
+		}
+		contents, readErr := migrationFiles.ReadFile(entry.Name())
+		require.NoError(t, readErr)
+		beforeMultiStage[entry.Name()] = &fstest.MapFile{Data: contents}
+	}
+	require.NoError(t, runMigrations(db, beforeMultiStage))
+
+	seedMigrationUser(ctx, t, db)
+	_, err = db.NewInsert().Model(&models.Workspace{ID: "workspace-1", Name: "Legacy"}).Exec(ctx)
+	require.NoError(t, err)
+	_, err = db.ExecContext(ctx, `INSERT INTO repost_policies (
+		id, workspace_id, name, enabled, delay_seconds, evaluation_window_seconds,
+		threshold_mode, created_by, updated_by
+	) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+		"policy-1", "workspace-1", "Legacy policy", true, 3600, 86400, "all", "user-1", "user-1")
+	require.NoError(t, err)
+	source := &models.SocialAccount{ID: "source-1", WorkspaceID: "workspace-1", Slug: "source", Platform: "x", AccountID: "source-provider", AccessTokenEnc: []byte("token"), IsActive: true}
+	target := &models.SocialAccount{ID: "target-1", WorkspaceID: "workspace-1", Slug: "target", Platform: "x", AccountID: "target-provider", AccessTokenEnc: []byte("token"), IsActive: true}
+	_, err = db.NewInsert().Model(source).Exec(ctx)
+	require.NoError(t, err)
+	_, err = db.NewInsert().Model(target).Exec(ctx)
+	require.NoError(t, err)
+	publication := &models.Publication{
+		ID: "publication-1", WorkspaceID: "workspace-1", CreatedByID: "user-1",
+		Intent: models.PublishingIntentPost, ContentProfile: models.ContentProfileShortText,
+		SourceContent: "Legacy post", Status: models.PublicationStatusPublished,
+	}
+	_, err = db.NewInsert().Model(publication).Exec(ctx)
+	require.NoError(t, err)
+	_, err = db.NewInsert().Model(&models.Rendition{
+		ID: "rendition-1", PublicationID: publication.ID, SocialAccountID: source.ID,
+		TargetKey: "x:source-provider", Platform: "x", Profile: models.ContentProfileShortText,
+		Status: models.RenditionStatusPublished, ExternalID: "source-post-1",
+	}).Exec(ctx)
+	require.NoError(t, err)
+	_, err = db.ExecContext(ctx, `INSERT INTO repost_executions (
+		id, workspace_id, publication_id, rendition_id, source_account_id, target_account_id,
+		policy_id, rule_snapshot_json, status, eligible_after, deadline_at, external_id
+	) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+		"execution-1", "workspace-1", publication.ID, "rendition-1", source.ID, target.ID,
+		"policy-1", `{"rule":{"delay_seconds":3600,"evaluation_window_seconds":86400,"threshold_mode":"all","plateau_checks":2}}`,
+		"succeeded", time.Now().UTC(), time.Now().UTC().Add(24*time.Hour), "legacy-repost-1")
+	require.NoError(t, err)
+
+	multiStageSQL, err := migrationFiles.ReadFile("132_multi_stage_reposts.sql")
+	require.NoError(t, err)
+	require.NoError(t, runMigrations(db, fstest.MapFS{
+		"132_multi_stage_reposts.sql": &fstest.MapFile{Data: multiStageSQL},
+	}))
+
+	var stagesJSON string
+	err = db.NewSelect().Table("repost_policies").Column("stages_json").Where("id = ?", "policy-1").Scan(ctx, &stagesJSON)
+	require.NoError(t, err)
+	require.Equal(t, "[]", stagesJSON)
+
+	var legacyExecution struct {
+		CurrentStage     int    `bun:"current_stage"`
+		TotalStages      int    `bun:"total_stages"`
+		UnrepostAttempts int    `bun:"unrepost_attempts"`
+		StageHistoryJSON string `bun:"stage_history_json"`
+		ExternalID       string `bun:"external_id"`
+	}
+	err = db.NewSelect().Table("repost_executions").
+		Column("current_stage", "total_stages", "unrepost_attempts", "stage_history_json", "external_id").
+		Where("id = ?", "execution-1").Scan(ctx, &legacyExecution)
+	require.NoError(t, err)
+	require.Equal(t, 1, legacyExecution.CurrentStage)
+	require.Equal(t, 1, legacyExecution.TotalStages)
+	require.Zero(t, legacyExecution.UnrepostAttempts)
+	require.Equal(t, "[]", legacyExecution.StageHistoryJSON)
+	require.Equal(t, "legacy-repost-1", legacyExecution.ExternalID)
+
+	for _, column := range []string{"current_stage", "total_stages", "unrepost_attempts", "stage_history_json"} {
+		exists, columnErr := migrationColumnExists(ctx, db, "repost_executions", column)
+		require.NoError(t, columnErr)
+		require.True(t, exists, "expected repost_executions.%s", column)
+	}
+}
+
 // isDuplicateColumnMigrationError keeps already-applied DDL idempotent during
 // upgrades; misclassifying a dialect error would either replay DDL or abort
 // the chain, so both dialect signatures are pinned.

backend/internal/models/models.go

@@ -1383,6 +1383,7 @@ type RepostPolicy struct {
 	MinViews                int64     `bun:"min_views,notnull,default:0" json:"min_views"`
 	RequirePlateau          bool      `bun:"require_plateau,notnull,default:false" json:"require_plateau"`
 	PlateauChecks           int       `bun:"plateau_checks,notnull,default:2" json:"plateau_checks"`
+	StagesJSON              string    `bun:"stages_json,notnull,default:'[]'" json:"stages_json"`
 	CreatedByID             string    `bun:"created_by,notnull" json:"created_by"`
 	UpdatedByID             string    `bun:"updated_by,notnull" json:"updated_by"`
 	CreatedAt               time.Time `bun:",nullzero,notnull,default:current_timestamp" json:"created_at"`
@@ -1430,6 +1431,10 @@ type RepostExecution struct {
 	PolicyID         string    `bun:"policy_id,nullzero" json:"policy_id,omitempty"`
 	RuleSnapshotJSON string    `bun:"rule_snapshot_json,notnull,default:'{}'" json:"rule_snapshot_json"`
 	Status           string    `bun:",notnull,default:'pending'" json:"status"`
+	CurrentStage     int       `bun:"current_stage,notnull,default:1" json:"current_stage"`
+	TotalStages      int       `bun:"total_stages,notnull,default:1" json:"total_stages"`
+	UnrepostAttempts int       `bun:"unrepost_attempts,notnull,default:0" json:"unrepost_attempts"`
+	StageHistoryJSON string    `bun:"stage_history_json,notnull,default:'[]'" json:"stage_history_json"`
 	EligibleAfter    time.Time `bun:"eligible_after,notnull" json:"eligible_after"`
 	DeadlineAt       time.Time `bun:"deadline_at,notnull" json:"deadline_at"`
 	NextCheckAt      time.Time `bun:"next_check_at,nullzero" json:"next_check_at"`

backend/internal/platform/bluesky.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/base64"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -786,6 +787,39 @@ func (b *BlueskyAdapter) Repost(ctx context.Context, accessToken, targetAccountI
 	return RepostResult{ExternalID: string(externalID), ExternalURL: req.ExternalURL}, nil
 }
 
+func (b *BlueskyAdapter) Unrepost(ctx context.Context, accessToken, targetAccountID string, req UnrepostRequest) error {
+	rkey := extractBlueskyRecordKey(req.RepostExternalID)
+	if strings.TrimSpace(targetAccountID) == "" || rkey == "" {
+		return fmt.Errorf("bluesky unrepost requires a target account and repost id")
+	}
+	_, err := b.doJSON(ctx, http.MethodPost, b.pdsURL+"/xrpc/com.atproto.repo.deleteRecord", map[string]any{
+		"repo": targetAccountID, "collection": "app.bsky.feed.repost", "rkey": rkey,
+	}, map[string]string{headerAuthorization: bearerPrefix + accessToken})
+	if err == nil {
+		return nil
+	}
+	var httpErr *HTTPError
+	if errors.As(err, &httpErr) && httpErr.Code == "RecordNotFound" {
+		return nil
+	}
+	return fmt.Errorf("unreposting on bluesky: %w", err)
+}
+
+func extractBlueskyRecordKey(externalID string) string {
+	trimmed := strings.TrimSpace(externalID)
+	if trimmed == "" {
+		return ""
+	}
+	var record struct {
+		URI string `json:"uri"`
+	}
+	if json.Unmarshal([]byte(trimmed), &record) == nil && record.URI != "" {
+		trimmed = record.URI
+	}
+	parts := strings.Split(strings.TrimRight(trimmed, "/"), "/")
+	return parts[len(parts)-1]
+}
+
 func (b *BlueskyAdapter) buildPostRecord(_ string, req *PublishRequest, createdAt time.Time) (map[string]interface{}, error) {
 	record := map[string]interface{}{
 		bskyRecordTypeField: "app.bsky.feed.post",

backend/internal/platform/bluesky_test.go

@@ -411,3 +411,35 @@ func TestBlueskyProviderKeyAdapterSharesAccountContentIdentity(t *testing.T) {
 	require.True(t, ok)
 	require.Equal(t, want, page.Items[0].ProviderContentID)
 }
+
+func TestBlueskyUnrepostOnlyIgnoresMissingRecord(t *testing.T) {
+	originalClient := httpClient
+	defer func() { httpClient = originalClient }()
+
+	tests := []struct {
+		name    string
+		status  int
+		body    string
+		wantErr bool
+	}{
+		{name: "missing record", status: http.StatusBadRequest, body: `{"error":"RecordNotFound"}`},
+		{name: "unrelated not found", status: http.StatusNotFound, body: `{"error":"NotFound"}`, wantErr: true},
+		{name: "other bad request", status: http.StatusBadRequest, body: `{"error":"InvalidRequest"}`, wantErr: true},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			httpClient = &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
+				return &http.Response{StatusCode: test.status, Header: make(http.Header), Body: io.NopCloser(strings.NewReader(test.body)), Request: req}, nil
+			})}
+			adapter := NewBlueskyAdapter(BlueskyDefaultPDSURL)
+			err := adapter.Unrepost(t.Context(), "token", "did:plc:target", UnrepostRequest{
+				RepostExternalID: `{"uri":"at://did:plc:target/app.bsky.feed.repost/record-key","cid":"cid"}`,
+			})
+			if test.wantErr {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+			}
+		})
+	}
+}

backend/internal/platform/errors.go

@@ -62,7 +62,7 @@ func providerErrorMetadata(body []byte) (string, string, string, string) {
 	if json.Unmarshal(body, &payload) != nil {
 		return "", "", "", ""
 	}
-	candidates := []any{payload["code"], payload["error_code"], payload["type"]}
+	candidates := []any{payload["code"], payload["error_code"], payload["type"], payload["error"]}
 	var subcodeCandidates []any
 	messageCandidates := []any{payload["message"]}
 	traceCandidates := []any{payload["fbtrace_id"], payload["trace_id"]}

backend/internal/platform/linkedin.go

@@ -672,6 +672,23 @@ func (l *LinkedInAdapter) Repost(ctx context.Context, accessToken, targetAccount
 	return RepostResult{ExternalID: headers.Get("x-restli-id"), ExternalURL: req.ExternalURL}, nil
 }
 
+func (l *LinkedInAdapter) Unrepost(ctx context.Context, accessToken, _ string, req UnrepostRequest) error {
+	urn := strings.TrimSpace(req.RepostExternalID)
+	if urn == "" {
+		return fmt.Errorf("linkedin unrepost requires a repost id")
+	}
+	_, err := DoRequest(ctx, http.MethodDelete, "https://api.linkedin.com/rest/posts/"+url.QueryEscape(urn), nil,
+		linkedinHeaders(accessToken, linkedInAPIVersion()))
+	if err == nil {
+		return nil
+	}
+	var httpErr *HTTPError
+	if errors.As(err, &httpErr) && (httpErr.StatusCode == http.StatusNotFound || httpErr.StatusCode == http.StatusGone) {
+		return nil
+	}
+	return fmt.Errorf("unreposting on linkedin: %w", err)
+}
+
 //nolint:gocyclo
 func (l *LinkedInAdapter) createPost(ctx context.Context, accessToken, authorURN, apiVersion string, req *PublishRequest) (string, error) {
 	visibility := firstNonEmptyString(settingString(req.Settings, "visibility"), "PUBLIC")

backend/internal/platform/linkedin_test.go

@@ -223,3 +223,19 @@ func TestLinkedInHideCommentUnsupported(t *testing.T) {
 		t.Fatalf("expected unsupported comment action, got %v", err)
 	}
 }
+
+func TestLinkedInUnrepostDeletesReshareURN(t *testing.T) {
+	originalClient := httpClient
+	defer func() { httpClient = originalClient }()
+	httpClient = &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
+		if req.Method != http.MethodDelete || req.URL.RequestURI() != "/rest/posts/urn%3Ali%3Ashare%3A12345" {
+			t.Fatalf("unexpected request %s %s", req.Method, req.URL.RequestURI())
+		}
+		return jsonResponseWithStatus(req, http.StatusNoContent, ""), nil
+	})}
+
+	adapter := NewLinkedInAdapter("", "", "", false)
+	if err := adapter.Unrepost(t.Context(), "token", "target", UnrepostRequest{RepostExternalID: "urn:li:share:12345"}); err != nil {
+		t.Fatalf("unrepost failed: %v", err)
+	}
+}

backend/internal/platform/mastodon.go

@@ -3,6 +3,7 @@ package platform
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"mime"
@@ -355,13 +356,35 @@ func (m *MastodonAdapter) Repost(ctx context.Context, accessToken, _ string, req
 		return RepostResult{}, fmt.Errorf("reposting on mastodon: %w", err)
 	}
 	var result struct {
-		ID  string `json:"id"`
-		URL string `json:"url"`
+		Reblog *struct {
+			ID string `json:"id"`
+		} `json:"reblog"`
 	}
 	if err := json.Unmarshal(body, &result); err != nil {
 		return RepostResult{}, fmt.Errorf("decoding mastodon repost: %w", err)
 	}
-	return RepostResult{ExternalID: result.ID, ExternalURL: result.URL}, nil
+	if result.Reblog != nil && strings.TrimSpace(result.Reblog.ID) != "" {
+		statusID = result.Reblog.ID
+	}
+	return RepostResult{ExternalID: statusID, ExternalURL: req.ExternalURL}, nil
+}
+
+func (m *MastodonAdapter) Unrepost(ctx context.Context, accessToken, _ string, req UnrepostRequest) error {
+	statusID := strings.TrimSpace(req.RepostExternalID)
+	if statusID == "" {
+		return fmt.Errorf("mastodon unrepost requires the target-local source status id")
+	}
+	_, err := DoRequest(ctx, http.MethodPost, m.instanceURL+"/api/v1/statuses/"+url.PathEscape(statusID)+"/unreblog", nil, map[string]string{
+		headerAuthorization: bearerPrefix + accessToken,
+	})
+	if err == nil {
+		return nil
+	}
+	var httpErr *HTTPError
+	if errors.As(err, &httpErr) && (httpErr.StatusCode == http.StatusNotFound || httpErr.StatusCode == http.StatusGone) {
+		return nil
+	}
+	return fmt.Errorf("unreposting on mastodon: %w", err)
 }
 
 func buildMastodonStatusForm(req *PublishRequest) (url.Values, error) {

backend/internal/platform/mastodon_test.go

@@ -8,6 +8,8 @@ import (
 	"net/url"
 	"strings"
 	"testing"
+
+	"github.com/stretchr/testify/require"
 )
 
 func TestMastodonResolveAccountPublishingCapabilitiesReadsInstanceVideoLimits(t *testing.T) {
@@ -133,3 +135,35 @@ func assertFormValue(t *testing.T, values url.Values, key, want string) {
 		t.Fatalf("expected %s=%q, got %q in %#v", key, want, got, values)
 	}
 }
+
+func TestMastodonRepostKeepsTargetLocalSourceIDForUnrepost(t *testing.T) {
+	originalClient := httpClient
+	defer func() { httpClient = originalClient }()
+
+	requests := 0
+	httpClient = &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
+		requests++
+		switch requests {
+		case 1:
+			require.Equal(t, "/api/v2/search", req.URL.Path)
+			return jsonResponse(req, `{"statuses":[{"id":"target-local-source"}]}`), nil
+		case 2:
+			require.Equal(t, "/api/v1/statuses/target-local-source/reblog", req.URL.Path)
+			return jsonResponse(req, `{"id":"reblog-wrapper","reblog":{"id":"target-local-source"}}`), nil
+		case 3:
+			require.Equal(t, "/api/v1/statuses/target-local-source/unreblog", req.URL.Path)
+			return jsonResponse(req, `{"id":"target-local-source","reblogged":false}`), nil
+		default:
+			t.Fatalf("unexpected request %d", requests)
+		}
+		return nil, nil
+	})}
+
+	adapter := NewMastodonAdapter("", "", "", "https://target.example")
+	result, err := adapter.Repost(t.Context(), "token", "target", RepostRequest{
+		SourceInstanceURL: "https://source.example", ExternalID: "source-instance-id", ExternalURL: "https://source.example/@author/1",
+	})
+	require.NoError(t, err)
+	require.Equal(t, "target-local-source", result.ExternalID)
+	require.NoError(t, adapter.Unrepost(t.Context(), "token", "target", UnrepostRequest{RepostExternalID: result.ExternalID}))
+}

backend/internal/platform/repost.go

@@ -16,8 +16,19 @@ type RepostResult struct {
 	ExternalURL string
 }
 
+type UnrepostRequest struct {
+	SourceExternalID string
+	RepostExternalID string
+}
+
 // RepostAdapter is an optional capability. Keeping it out of Adapter lets
 // providers without a native repost API remain valid publishing adapters.
 type RepostAdapter interface {
 	Repost(ctx context.Context, accessToken, targetAccountID string, req RepostRequest) (RepostResult, error)
 }
+
+// UnrepostAdapter is optional because a provider may support native reposts
+// without offering a safe way to remove one later.
+type UnrepostAdapter interface {
+	Unrepost(ctx context.Context, accessToken, targetAccountID string, req UnrepostRequest) error
+}