+ ${d.postizStatus === 'published'
+ ? `\u2713 Published${d.publishedAt ? ' ' + new Date(d.publishedAt).toLocaleString('en-US', { month: 'short', day: 'numeric', hour: 'numeric', minute: '2-digit' }) : ''}${d.postizReleaseURL ? ` —
open` : ''}`
+ : d.postizStatus === 'failed'
+ ? `\u26a0 Postiz error: ${esc(d.postizError || 'unknown')}`
+ : `\u23f3 Queued in Postiz (${esc(d.postizPostId || '')})`}
+
${icon}
@@ -1339,7 +1397,11 @@ class FolkCampaignPlanner extends HTMLElement {
const cards = items.map(node => {
const d = node.data as PostNodeData;
- const statusColor = d.status === 'published' ? '#22c55e' : d.status === 'scheduled' ? '#3b82f6' : '#f59e0b';
+ const statusColor = d.postizStatus === 'published' || d.status === 'published' ? '#22c55e'
+ : d.postizStatus === 'failed' ? '#ef4444'
+ : d.postizStatus === 'queued' ? '#a855f7'
+ : d.status === 'scheduled' ? '#3b82f6'
+ : '#f59e0b';
const dateStr = d.scheduledAt ? new Date(d.scheduledAt).toLocaleDateString('en-US', { month: 'short', day: 'numeric' }) : 'Unscheduled';
const preview = (d.content || '').split('\n')[0].substring(0, 60);
const tags = d.hashtags.slice(0, 3).map(h =>
@@ -1396,7 +1458,11 @@ class FolkCampaignPlanner extends HTMLElement {
const platform = d.platform || 'x';
const color = PLATFORM_COLORS[platform] || '#888';
const icon = PLATFORM_ICONS[platform] || platform.charAt(0);
- const statusColor = d.status === 'published' ? '#22c55e' : d.status === 'scheduled' ? '#3b82f6' : '#f59e0b';
+ const statusColor = d.postizStatus === 'published' || d.status === 'published' ? '#22c55e'
+ : d.postizStatus === 'failed' ? '#ef4444'
+ : d.postizStatus === 'queued' ? '#a855f7'
+ : d.status === 'scheduled' ? '#3b82f6'
+ : '#f59e0b';
const statusLabel = d.status ? d.status.charAt(0).toUpperCase() + d.status.slice(1) : 'Draft';
const dateStr = d.scheduledAt ? new Date(d.scheduledAt).toLocaleDateString('en-US', { month: 'short', day: 'numeric', hour: 'numeric', minute: '2-digit' }) : '—';
const preview = (d.content || '').split('\n')[0].substring(0, 50);
@@ -1627,7 +1693,11 @@ class FolkCampaignPlanner extends HTMLElement {
const platform = d.platform || 'x';
const color = PLATFORM_COLORS[platform] || '#888';
const icon = PLATFORM_ICONS[platform] || platform.charAt(0);
- const statusColor = d.status === 'published' ? '#22c55e' : d.status === 'scheduled' ? '#3b82f6' : '#f59e0b';
+ const statusColor = d.postizStatus === 'published' || d.status === 'published' ? '#22c55e'
+ : d.postizStatus === 'failed' ? '#ef4444'
+ : d.postizStatus === 'queued' ? '#a855f7'
+ : d.status === 'scheduled' ? '#3b82f6'
+ : '#f59e0b';
const preview = (d.content || '').split('\n')[0].substring(0, 50);
const charCount = (d.content || '').length;
const charMax = platform === 'x' ? 280 : 2200;
diff --git a/modules/rsocials/lib/postiz-client.ts b/modules/rsocials/lib/postiz-client.ts
index b249e82a..18b6b22f 100644
--- a/modules/rsocials/lib/postiz-client.ts
+++ b/modules/rsocials/lib/postiz-client.ts
@@ -28,19 +28,26 @@ export async function getPostizConfig(spaceSlug: string): Promise
`). The API lives at `/api/public/v1/*`; we
+ * tolerate either `config.url = 'https://host'` or `config.url = 'https://host/api'`
+ * by inserting `/api` when missing.
+ */
export async function postizFetch(
config: PostizConfig,
path: string,
opts: RequestInit = {},
): Promise {
const headers = new Headers(opts.headers);
- headers.set("Authorization", `Bearer ${config.apiKey}`);
+ headers.set("Authorization", config.apiKey);
if (!headers.has("Content-Type") && opts.body) {
headers.set("Content-Type", "application/json");
}
- return fetch(`${config.url}${path}`, { ...opts, headers });
+ const base = config.url.endsWith('/api') ? config.url : `${config.url}/api`;
+ return fetch(`${base}${path}`, { ...opts, headers });
}
/** GET /public/v1/integrations — list connected social channels. */
@@ -50,20 +57,70 @@ export async function getIntegrations(config: PostizConfig) {
return res.json();
}
-/** POST /public/v1/posts — create a single post (draft, schedule, or now). */
+/** POST /public/v1/posts — create a single post (draft, schedule, or now).
+ *
+ * Postiz expects a nested shape:
+ * { type, date, shortLink, tags, posts: [{ integration: {id}, value: [{content}], settings: {__type, ...} }] }
+ *
+ * Callers pass integrations as {id, identifier} tuples; identifier populates
+ * settings.__type which Postiz uses to route to the correct provider handler.
+ */
+export interface PostizIntegrationRef {
+ id: string;
+ identifier: string; // e.g. 'x', 'linkedin', 'bluesky'
+}
+
+// Platform-specific settings Postiz validates on POST /posts. Keys that are
+// required-non-empty for each provider live here. We merge defaults when the
+// caller doesn't supply overrides.
+function defaultSettingsFor(identifier: string): Record {
+ const base: Record = { __type: identifier };
+ switch (identifier) {
+ case 'x':
+ base.who_can_reply_post = 'everyone';
+ break;
+ case 'linkedin':
+ base.post_to_company = false;
+ break;
+ case 'instagram':
+ base.post_type = 'post';
+ break;
+ case 'youtube':
+ base.type = 'public';
+ base.title = '';
+ base.category = '22';
+ break;
+ // Others (bluesky, threads, mastodon, reddit, ...) don't require settings
+ // beyond __type as of Postiz 5.x.
+ }
+ return base;
+}
+
export async function createPost(
config: PostizConfig,
payload: {
content: string;
- integrationIds: string[];
+ integrations: PostizIntegrationRef[];
type: 'draft' | 'schedule' | 'now';
scheduledAt?: string;
group?: string;
},
) {
+ const body = {
+ type: payload.type,
+ date: payload.scheduledAt || new Date().toISOString(),
+ shortLink: false,
+ tags: [] as unknown[],
+ posts: payload.integrations.map(integ => ({
+ integration: { id: integ.id },
+ value: [{ content: payload.content, image: [] as unknown[] }],
+ ...(payload.group ? { group: payload.group } : {}),
+ settings: defaultSettingsFor(integ.identifier),
+ })),
+ };
const res = await postizFetch(config, "/public/v1/posts", {
method: "POST",
- body: JSON.stringify(payload),
+ body: JSON.stringify(body),
});
if (!res.ok) {
const text = await res.text().catch(() => '');
@@ -72,27 +129,66 @@ export async function createPost(
return res.json();
}
-/** Create a thread — sends multiple grouped posts sharing a group ID. */
+/** GET /public/v1/posts — list posts in a date range (used for status reconciliation). */
+export interface PostizListedPost {
+ id: string;
+ content?: string;
+ publishDate?: string;
+ releaseURL?: string;
+ state: 'QUEUE' | 'PUBLISHED' | 'ERROR' | 'DRAFT';
+ integration?: { id: string; providerIdentifier?: string; name?: string };
+}
+
+export async function listPosts(
+ config: PostizConfig,
+ startDate: Date,
+ endDate: Date,
+): Promise {
+ const qs = new URLSearchParams({
+ startDate: startDate.toISOString(),
+ endDate: endDate.toISOString(),
+ });
+ const res = await postizFetch(config, `/public/v1/posts?${qs}`);
+ if (!res.ok) throw new Error(`Postiz listPosts error: ${res.status}`);
+ const data = await res.json();
+ // Postiz wraps list as { posts: [...] }; be defensive against raw array.
+ if (Array.isArray(data)) return data as PostizListedPost[];
+ if (Array.isArray(data?.posts)) return data.posts as PostizListedPost[];
+ return [];
+}
+
+/** Create a thread — single Postiz post whose `value` array carries each tweet.
+ *
+ * Postiz treats the `value` array on a PostItem as the thread segments, so we
+ * send ONE request with multiple value entries, not N grouped requests.
+ */
export async function createThread(
config: PostizConfig,
tweets: string[],
opts: {
- integrationIds: string[];
+ integrations: PostizIntegrationRef[];
type: 'draft' | 'schedule' | 'now';
scheduledAt?: string;
},
) {
- const group = `thread-${Date.now()}-${Math.random().toString(36).substring(2, 8)}`;
- const results = [];
- for (const content of tweets) {
- const result = await createPost(config, {
- content,
- integrationIds: opts.integrationIds,
- type: opts.type,
- scheduledAt: opts.scheduledAt,
- group,
- });
- results.push(result);
+ const body = {
+ type: opts.type,
+ date: opts.scheduledAt || new Date().toISOString(),
+ shortLink: false,
+ tags: [] as unknown[],
+ posts: opts.integrations.map(integ => ({
+ integration: { id: integ.id },
+ value: tweets.map(t => ({ content: t, image: [] as unknown[] })),
+ settings: defaultSettingsFor(integ.identifier),
+ })),
+ };
+ const res = await postizFetch(config, "/public/v1/posts", {
+ method: "POST",
+ body: JSON.stringify(body),
+ });
+ if (!res.ok) {
+ const text = await res.text().catch(() => '');
+ throw new Error(`Postiz createThread error: ${res.status} ${text}`);
}
- return { group, posts: results };
+ return res.json();
}
diff --git a/modules/rsocials/mod.ts b/modules/rsocials/mod.ts
index e79754fb..84d7093a 100644
--- a/modules/rsocials/mod.ts
+++ b/modules/rsocials/mod.ts
@@ -31,16 +31,21 @@ import {
} from "./lib/image-gen";
import { DEMO_FEED } from "./lib/types";
import { getListmonkConfig, listmonkFetch } from "./lib/listmonk-proxy";
-import { getPostizConfig, getIntegrations, createPost, createThread } from "./lib/postiz-client";
+import { getPostizConfig, getIntegrations, createPost, createThread, listPosts, type PostizIntegrationRef } from "./lib/postiz-client";
import { verifyToken, extractToken } from "../../server/auth";
import type { EncryptIDClaims } from "../../server/auth";
import { resolveCallerRole, roleAtLeast } from "../../server/spaces";
import type { SpaceRoleString } from "../../server/spaces";
let _syncServer: SyncServer | null = null;
+let _postizSweepTimer: ReturnType | null = null;
const routes = new Hono();
+// Sweep the postiz lead window: anything with scheduledAt within this many ms
+// and not yet pushed to Postiz will be sent. Past-due posts push as type:'now'.
+const POSTIZ_SWEEP_LEAD_MS = 10 * 60 * 1000; // 10 minutes
+
// ── Automerge doc management ──
function ensureDoc(space: string): SocialsDoc {
@@ -785,9 +790,16 @@ routes.post("/api/postiz/posts", async (c) => {
}
try {
+ const fetched = await getIntegrations(config);
+ const byId = new Map((fetched || []).map((i: any) => [i.id, i]));
+ const integrations: PostizIntegrationRef[] = integrationIds
+ .map((id: string) => byId.get(id))
+ .filter(Boolean)
+ .map((i: any) => ({ id: i.id, identifier: i.identifier || i.providerIdentifier }));
+ if (!integrations.length) return c.json({ error: "No integration matched the requested IDs" }, 400);
const result = await createPost(config, {
content,
- integrationIds,
+ integrations,
type: type || 'draft',
scheduledAt,
});
@@ -797,6 +809,281 @@ routes.post("/api/postiz/posts", async (c) => {
}
});
+// Resolve an integration id from a platform name against a live Postiz integrations list.
+// Resolve a full integration tuple {id, identifier} from a platform name.
+// Postiz's public API returns integrations with `identifier` (e.g. 'x'), while
+// the internal/admin shape uses `providerIdentifier`. Exact match on either
+// beats a loose name includes().
+function resolveIntegration(integrations: any[], platform: string): PostizIntegrationRef | null {
+ const toRef = (i: any): PostizIntegrationRef | null => {
+ const id = i?.id;
+ const identifier = i?.identifier || i?.providerIdentifier;
+ return id && identifier ? { id, identifier } : null;
+ };
+ if (!platform) return toRef(integrations[0]);
+ const needle = platform.toLowerCase();
+ const exact = integrations.find((i: any) =>
+ i.identifier?.toLowerCase() === needle ||
+ i.providerIdentifier?.toLowerCase() === needle
+ );
+ if (exact) return toRef(exact);
+ const loose = integrations.find((i: any) => i.name?.toLowerCase().includes(needle));
+ return toRef(loose) || toRef(integrations[0]);
+}
+
+// Push a single campaign-flow Post node to Postiz. Writes postizPostId + status
+// back onto the node via Automerge so every client sees the update.
+async function sendCampaignNodeToPostiz(
+ space: string,
+ flowId: string,
+ nodeId: string,
+): Promise<{ ok: true; postizPostId: string } | { ok: false; error: string; code: number }> {
+ const config = await getPostizConfig(space);
+ if (!config) return { ok: false, error: "Postiz not configured for this space", code: 404 };
+
+ const docId = socialsDocId(space);
+ const doc = _syncServer!.getDoc(docId);
+ const flow = doc?.campaignFlows?.[flowId];
+ if (!flow) return { ok: false, error: "Flow not found", code: 404 };
+ const node = flow.nodes.find(n => n.id === nodeId);
+ if (!node || node.type !== 'post') return { ok: false, error: "Post node not found", code: 404 };
+
+ const data = node.data as PostNodeData;
+ if (data.postizPostId) return { ok: false, error: "Already sent to Postiz", code: 409 };
+
+ const content = (data.content || '').trim();
+ if (!content) return { ok: false, error: "Post has no content", code: 400 };
+
+ let integrations: PostizIntegrationRef[];
+ try {
+ const fetched = await getIntegrations(config);
+ if (!Array.isArray(fetched) || fetched.length === 0) {
+ return { ok: false, error: "No Postiz integrations configured", code: 400 };
+ }
+ let ref: PostizIntegrationRef | null = null;
+ if (data.postizIntegrationId) {
+ const matched = fetched.find((i: any) => i.id === data.postizIntegrationId);
+ if (matched) ref = { id: matched.id, identifier: matched.identifier || matched.providerIdentifier };
+ }
+ ref = ref || resolveIntegration(fetched, data.platform);
+ if (!ref) return { ok: false, error: `No integration matches platform "${data.platform}"`, code: 400 };
+ integrations = [ref];
+ } catch (err: any) {
+ return { ok: false, error: `Postiz integrations fetch failed: ${err.message}`, code: 502 };
+ }
+
+ const scheduledAt = data.scheduledAt ? new Date(data.scheduledAt) : null;
+ const now = new Date();
+ const type: 'schedule' | 'now' = !scheduledAt || scheduledAt.getTime() <= now.getTime() ? 'now' : 'schedule';
+ const hashtagLine = data.hashtags?.length ? '\n\n' + data.hashtags.map(h => h.startsWith('#') ? h : `#${h}`).join(' ') : '';
+ const payload = {
+ content: content + hashtagLine,
+ integrations,
+ type,
+ scheduledAt: type === 'schedule' ? scheduledAt!.toISOString() : undefined,
+ };
+
+ try {
+ const result = await createPost(config, payload);
+ // Postiz returns either { id } or an array — be defensive on parse.
+ let postizPostId: string = (result as any)?.id
+ || (Array.isArray(result) && (result[0] as any)?.id)
+ || (result as any)?.posts?.[0]?.id
+ || '';
+
+ // Fallback: Postiz's create response shape varies by version and often
+ // doesn't include the new post id. Query listPosts in a narrow window,
+ // then find by integration + content to recover the real id.
+ if (!postizPostId) {
+ try {
+ const windowStart = new Date(Date.now() - 60_000);
+ const windowEnd = new Date(Date.now() + 365 * 86400_000);
+ const listed = await listPosts(config, windowStart, windowEnd);
+ const match = listed.find(p =>
+ p.integration?.id === integrations[0].id &&
+ (p.content || '').trim() === payload.content.trim()
+ );
+ if (match) postizPostId = match.id;
+ } catch { /* leave id empty, reconcile will treat as lost */ }
+ }
+
+ _syncServer!.changeDoc(docId, `postiz send ${nodeId}`, (d) => {
+ const f = d.campaignFlows?.[flowId];
+ if (!f) return;
+ const n = f.nodes.find(x => x.id === nodeId);
+ if (!n || n.type !== 'post') return;
+ const nd = n.data as PostNodeData;
+ nd.postizPostId = postizPostId || `postiz-${Date.now()}`;
+ nd.postizIntegrationId = integrations[0].id;
+ nd.postizStatus = 'queued';
+ nd.postizSentAt = Date.now();
+ nd.postizError = '';
+ if (nd.status === 'draft') nd.status = 'scheduled';
+ f.updatedAt = Date.now();
+ });
+
+ return { ok: true, postizPostId };
+ } catch (err: any) {
+ _syncServer!.changeDoc(docId, `postiz send failed ${nodeId}`, (d) => {
+ const f = d.campaignFlows?.[flowId];
+ const n = f?.nodes.find(x => x.id === nodeId);
+ if (!n || n.type !== 'post') return;
+ const nd = n.data as PostNodeData;
+ nd.postizStatus = 'failed';
+ nd.postizError = err.message || 'Unknown error';
+ });
+ return { ok: false, error: err.message || 'Postiz createPost failed', code: 502 };
+ }
+}
+
+// Periodic sweep: for every rsocials doc, find post nodes with scheduledAt
+// approaching (within lead window) or already past, status === 'scheduled',
+// and no postizPostId yet — push them to Postiz.
+async function postizSweep() {
+ if (!_syncServer) return;
+ const allDocIds = _syncServer.listDocs?.() || [];
+ const now = Date.now();
+ for (const docId of allDocIds) {
+ if (!docId.endsWith(':socials:data')) continue;
+ const space = docId.split(':')[0];
+ if (!space) continue;
+
+ let config;
+ try { config = await getPostizConfig(space); } catch { continue; }
+ if (!config) continue; // Skip spaces without Postiz wired up
+
+ const doc = _syncServer.getDoc(docId);
+ if (!doc?.campaignFlows) continue;
+
+ for (const flow of Object.values(doc.campaignFlows)) {
+ for (const node of flow.nodes) {
+ if (node.type !== 'post') continue;
+ const d = node.data as PostNodeData;
+ if (d.postizPostId) continue;
+ // Skip any node that's already been handled by Postiz (queued/published/failed).
+ // A failed send needs the user to clear postizStatus before we retry.
+ if (d.postizStatus) continue;
+ if (d.status !== 'scheduled') continue;
+ if (!d.scheduledAt) continue;
+ const ts = new Date(d.scheduledAt).getTime();
+ if (Number.isNaN(ts)) continue;
+ if (ts - now > POSTIZ_SWEEP_LEAD_MS) continue; // Too far out — wait
+
+ try {
+ const res = await sendCampaignNodeToPostiz(space, flow.id, node.id);
+ if (res.ok) {
+ console.log(`[rsocials:postiz-sweep] sent node=${node.id} space=${space} postiz=${res.postizPostId}`);
+ } else {
+ console.warn(`[rsocials:postiz-sweep] skip node=${node.id} space=${space} reason=${res.error}`);
+ }
+ } catch (err: any) {
+ console.error(`[rsocials:postiz-sweep] error node=${node.id}: ${err.message}`);
+ }
+ }
+ }
+ }
+}
+
+// Reconcile local queued posts against Postiz. For each space with Postiz
+// wired up, query posts in the window covering the oldest queued node, then
+// flip statuses to 'published' or 'failed' based on Postiz state.
+async function postizReconcile() {
+ if (!_syncServer) return;
+ const allDocIds = _syncServer.listDocs?.() || [];
+ const now = Date.now();
+
+ for (const docId of allDocIds) {
+ if (!docId.endsWith(':socials:data')) continue;
+ const space = docId.split(':')[0];
+ if (!space) continue;
+
+ // Collect queued nodes first — skip the API call if nothing to reconcile.
+ const doc = _syncServer.getDoc(docId);
+ if (!doc?.campaignFlows) continue;
+
+ const queuedRefs: { flowId: string; nodeId: string; postizPostId: string; sentAt: number }[] = [];
+ for (const flow of Object.values(doc.campaignFlows)) {
+ for (const node of flow.nodes) {
+ if (node.type !== 'post') continue;
+ const d = node.data as PostNodeData;
+ if (!d.postizPostId) continue;
+ if (d.postizStatus !== 'queued') continue;
+ queuedRefs.push({
+ flowId: flow.id,
+ nodeId: node.id,
+ postizPostId: d.postizPostId,
+ sentAt: d.postizSentAt || now,
+ });
+ }
+ }
+ if (queuedRefs.length === 0) continue;
+
+ let config;
+ try { config = await getPostizConfig(space); } catch { continue; }
+ if (!config) continue;
+
+ // Query window: oldest queued send minus 1h, through 30 days out.
+ const oldest = Math.min(...queuedRefs.map(q => q.sentAt));
+ const startDate = new Date(oldest - 60 * 60 * 1000);
+ const endDate = new Date(now + 30 * 24 * 60 * 60 * 1000);
+
+ let listed: Awaited>;
+ try {
+ listed = await listPosts(config, startDate, endDate);
+ } catch (err: any) {
+ console.warn(`[rsocials:postiz-reconcile] list failed space=${space}: ${err.message}`);
+ continue;
+ }
+ const byId = new Map(listed.map(p => [p.id, p]));
+
+ _syncServer.changeDoc(docId, `postiz reconcile ${space}`, (d) => {
+ for (const ref of queuedRefs) {
+ const remote = byId.get(ref.postizPostId);
+ const flow = d.campaignFlows?.[ref.flowId];
+ const node = flow?.nodes.find(n => n.id === ref.nodeId);
+ if (!node || node.type !== 'post') continue;
+ const nd = node.data as PostNodeData;
+ nd.postizCheckedAt = now;
+ if (!remote) continue; // Postiz doesn't know the ID — maybe out of window
+ if (remote.state === 'PUBLISHED') {
+ nd.postizStatus = 'published';
+ nd.status = 'published';
+ nd.publishedAt = remote.publishDate ? Date.parse(remote.publishDate) : now;
+ if (remote.releaseURL) nd.postizReleaseURL = remote.releaseURL;
+ } else if (remote.state === 'ERROR') {
+ nd.postizStatus = 'failed';
+ nd.postizError = 'Postiz reported publish error';
+ }
+ // QUEUE / DRAFT: leave as-is.
+ }
+ });
+ }
+}
+
+function startPostizScheduler() {
+ if (_postizSweepTimer) return;
+ _postizSweepTimer = setInterval(() => {
+ postizSweep().catch(() => {});
+ postizReconcile().catch(() => {});
+ }, 60_000);
+ setTimeout(() => {
+ postizSweep().catch(() => {});
+ postizReconcile().catch(() => {});
+ }, 30_000);
+ console.log('[rsocials] Postiz scheduler sweep started (60s interval, 10min lead, status reconcile)');
+}
+
+routes.post("/api/campaign/flows/:flowId/nodes/:nodeId/send-postiz", async (c) => {
+ const space = c.req.param("space") || "demo";
+ const dataSpace = c.get("effectiveSpace") || space;
+ const flowId = c.req.param("flowId");
+ const nodeId = c.req.param("nodeId");
+
+ const result = await sendCampaignNodeToPostiz(dataSpace, flowId, nodeId);
+ if (!result.ok) return c.json({ error: result.error }, result.code as any);
+ return c.json({ ok: true, postizPostId: result.postizPostId });
+});
+
routes.post("/api/postiz/threads", async (c) => {
const space = c.req.param("space") || "demo";
const config = await getPostizConfig(space);
@@ -809,20 +1096,25 @@ routes.post("/api/postiz/threads", async (c) => {
}
// If no integrationIds provided, try to auto-detect from configured integrations
- let ids = integrationIds;
- if (!ids?.length) {
- try {
- const integrations = await getIntegrations(config);
- ids = (integrations || []).map((i: any) => i.id).slice(0, 1);
- } catch { /* fall through */ }
- }
- if (!ids?.length) {
- return c.json({ error: "No integrationIds provided and no integrations found" }, 400);
+ let integrations: PostizIntegrationRef[] = [];
+ try {
+ const fetched = await getIntegrations(config);
+ const byId = new Map((fetched || []).map((i: any) => [i.id, i]));
+ const ids: string[] = Array.isArray(integrationIds) && integrationIds.length
+ ? integrationIds
+ : (fetched || []).slice(0, 1).map((i: any) => i.id);
+ integrations = ids
+ .map((id: string) => byId.get(id))
+ .filter(Boolean)
+ .map((i: any) => ({ id: i.id, identifier: i.identifier || i.providerIdentifier }));
+ } catch { /* fall through */ }
+ if (!integrations.length) {
+ return c.json({ error: "No integrations provided and no integrations found" }, 400);
}
try {
const result = await createThread(config, tweets, {
- integrationIds: ids,
+ integrations,
type: type || 'draft',
scheduledAt,
});
@@ -1176,29 +1468,24 @@ routes.post("/api/campaign-workflows/:id/run", async (c) => {
switch (node.type) {
case 'post-to-platform': {
if (!postizConfig) throw new Error('Postiz not configured');
- const integrations = await getIntegrations(postizConfig);
- const platformName = (cfg.platform as string || '').toLowerCase();
- const match = (integrations || []).find((i: any) =>
- i.name?.toLowerCase().includes(platformName) || i.providerIdentifier?.toLowerCase().includes(platformName)
- );
- const integrationIds = match ? [match.id] : (integrations || []).slice(0, 1).map((i: any) => i.id);
+ const fetched = await getIntegrations(postizConfig);
+ const ref = resolveIntegration(fetched || [], (cfg.platform as string) || '');
+ if (!ref) throw new Error(`No integration matches platform "${cfg.platform}"`);
const content = (cfg.content as string || '') + (cfg.hashtags ? '\n' + cfg.hashtags : '');
- await createPost(postizConfig, { content, integrationIds, type: 'draft' });
+ await createPost(postizConfig, { content, integrations: [ref], type: 'draft' });
results.push({ nodeId: node.id, status: 'success', message: `Draft created on ${cfg.platform || 'default'}`, durationMs: Date.now() - start });
break;
}
case 'cross-post': {
if (!postizConfig) throw new Error('Postiz not configured');
const platforms = (cfg.platforms as string || '').split(',').map(s => s.trim().toLowerCase()).filter(Boolean);
- const integrations = await getIntegrations(postizConfig);
+ const fetched = await getIntegrations(postizConfig);
const content = (cfg.content as string || '');
const posted: string[] = [];
for (const plat of platforms) {
- const match = (integrations || []).find((i: any) =>
- i.name?.toLowerCase().includes(plat) || i.providerIdentifier?.toLowerCase().includes(plat)
- );
- if (match) {
- await createPost(postizConfig, { content, integrationIds: [match.id], type: 'draft' });
+ const ref = resolveIntegration(fetched || [], plat);
+ if (ref) {
+ await createPost(postizConfig, { content, integrations: [ref], type: 'draft' });
posted.push(plat);
}
}
@@ -1216,13 +1503,10 @@ routes.post("/api/campaign-workflows/:id/run", async (c) => {
}
const tweets = threadContent.split(/\n---\n/).map(s => s.trim()).filter(Boolean);
if (tweets.length === 0) throw new Error('No thread content or thread ID provided');
- const integrations = await getIntegrations(postizConfig);
- const platformName = (cfg.platform as string || '').toLowerCase();
- const match = (integrations || []).find((i: any) =>
- i.name?.toLowerCase().includes(platformName) || i.providerIdentifier?.toLowerCase().includes(platformName)
- );
- const integrationIds = match ? [match.id] : (integrations || []).slice(0, 1).map((i: any) => i.id);
- await createThread(postizConfig, tweets, { integrationIds, type: 'draft' });
+ const fetched = await getIntegrations(postizConfig);
+ const ref = resolveIntegration(fetched || [], (cfg.platform as string) || '');
+ if (!ref) throw new Error(`No integration matches platform "${cfg.platform}"`);
+ await createThread(postizConfig, tweets, { integrations: [ref], type: 'draft' });
results.push({ nodeId: node.id, status: 'success', message: `Thread draft created (${tweets.length} tweets)`, durationMs: Date.now() - start });
break;
}
@@ -2514,8 +2798,8 @@ routes.get("/campaign-flow", (c) => {
modules: getModuleInfoList(),
theme: "dark",
body: ``,
- styles: ``,
- scripts: ``,
+ styles: ``,
+ scripts: ``,
}));
});
@@ -2879,6 +3163,7 @@ export const socialsModule: RSpaceModule = {
_syncServer = ctx.syncServer;
// Run migration for any existing file-based threads
try { await migrateFileThreadsToAutomerge("demo"); } catch { /* ignore */ }
+ startPostizScheduler();
},
externalApp: { url: POSTIZ_URL, name: "Postiz" },
feeds: [
diff --git a/modules/rsocials/schemas.ts b/modules/rsocials/schemas.ts
index d7166003..d1f71ed8 100644
--- a/modules/rsocials/schemas.ts
+++ b/modules/rsocials/schemas.ts
@@ -66,6 +66,14 @@ export interface PostNodeData {
scheduledAt: string;
status: 'draft' | 'scheduled' | 'published';
hashtags: string[];
+ postizPostId?: string;
+ postizIntegrationId?: string;
+ postizStatus?: 'queued' | 'published' | 'failed';
+ postizError?: string;
+ postizSentAt?: number;
+ postizCheckedAt?: number;
+ postizReleaseURL?: string;
+ publishedAt?: number;
}
export interface ThreadNodeData {
diff --git a/shared/collab-presence.ts b/shared/collab-presence.ts
index 975859fb..f1f12058 100644
--- a/shared/collab-presence.ts
+++ b/shared/collab-presence.ts
@@ -57,6 +57,7 @@ export function broadcastPresence(opts: PresenceOpts): void {
noteId: opts.noteId,
itemId: opts.itemId,
username: session.username,
+ userId: session.userId,
color: userColor(session.userId),
});
}
@@ -75,21 +76,17 @@ export function broadcastLeave(): void {
}
/**
- * Start a 10-second heartbeat that broadcasts presence.
+ * Start a 5-second heartbeat that broadcasts presence.
* Returns a cleanup function to stop the heartbeat.
*
+ * No beforeunload leave — navigating between rApps in the same space would
+ * otherwise flash the user offline to peers. Peers GC us after ~12s if we
+ * truly close the tab.
+ *
* @param getOpts - Called each heartbeat to get current presence state
*/
export function startPresenceHeartbeat(getOpts: () => PresenceOpts): () => void {
broadcastPresence(getOpts());
- const timer = setInterval(() => broadcastPresence(getOpts()), 10_000);
-
- // Clean up immediately on page unload
- const onUnload = () => broadcastLeave();
- window.addEventListener('beforeunload', onUnload);
-
- return () => {
- clearInterval(timer);
- window.removeEventListener('beforeunload', onUnload);
- };
+ const timer = setInterval(() => broadcastPresence(getOpts()), 5_000);
+ return () => clearInterval(timer);
}
diff --git a/shared/components/rstack-collab-overlay.ts b/shared/components/rstack-collab-overlay.ts
index 97e4f0bd..2ccb6fb9 100644
--- a/shared/components/rstack-collab-overlay.ts
+++ b/shared/components/rstack-collab-overlay.ts
@@ -27,6 +27,7 @@ const PEER_COLORS = [
interface PeerState {
peerId: string;
+ userId?: string; // stable identity across page nav / peerId churn
username: string;
color: string;
cursor: { x: number; y: number } | null;
@@ -242,16 +243,25 @@ export class RStackCollabOverlay extends HTMLElement {
this.#connectToDoc();
}
- // Listen for space-wide presence broadcasts (module context from all rApps)
+ // Listen for space-wide presence broadcasts (module context from all rApps).
+ // Key by userId when available so a peer navigating between rApps (new WS peerId
+ // each page) maps back to the same row instead of flashing offline.
this.#unsubPresence = runtime.onCustomMessage('presence', (msg: any) => {
if (!msg.username) return;
- // Use peerId from server relay, or fall back to a hash of username
const pid = msg.peerId || `anon-${msg.username}`;
if (pid === this.#localPeerId) return; // ignore self
- const existing = this.#peers.get(pid);
- this.#peers.set(pid, {
+ const key = msg.userId || pid;
+ // Drop any stale entry keyed by the old peerId for this same userId
+ if (msg.userId) {
+ for (const [k, p] of this.#peers) {
+ if (k !== key && p.userId === msg.userId) this.#peers.delete(k);
+ }
+ }
+ const existing = this.#peers.get(key);
+ this.#peers.set(key, {
peerId: pid,
+ userId: msg.userId || existing?.userId,
username: msg.username || existing?.username || 'Anonymous',
color: msg.color || existing?.color || this.#colorForPeer(pid),
cursor: existing?.cursor ?? null,
@@ -499,7 +509,7 @@ export class RStackCollabOverlay extends HTMLElement {
#gcPeers() {
const now = Date.now();
- const staleThreshold = this.#externalPeers ? 30000 : 15000;
+ const staleThreshold = this.#externalPeers ? 30000 : 12000;
let changed = false;
for (const [id, peer] of this.#peers) {
if (now - peer.lastSeen > staleThreshold) {
@@ -522,13 +532,27 @@ export class RStackCollabOverlay extends HTMLElement {
}
}
- /** Deduplicate peers by username, keeping the most recently seen entry per user. */
+ /**
+ * Deduplicate peers for display. Keyed by username (lowercase) so a user
+ * appearing twice — e.g. old awareness entry (peerId-keyed) alongside a new
+ * presence entry (userId-keyed) after navigating rApps — collapses to one
+ * row. Prefers entries carrying module/context metadata so the panel shows
+ * which rApp the peer is in.
+ */
#uniquePeers(): PeerState[] {
const byName = new Map();
for (const peer of this.#peers.values()) {
const key = peer.username.toLowerCase();
const existing = byName.get(key);
- if (!existing || peer.lastSeen > existing.lastSeen) {
+ if (!existing) {
+ byName.set(key, peer);
+ continue;
+ }
+ const peerHasMeta = !!(peer.module || peer.context);
+ const existingHasMeta = !!(existing.module || existing.context);
+ if (peerHasMeta && !existingHasMeta) {
+ byName.set(key, peer);
+ } else if (peerHasMeta === existingHasMeta && peer.lastSeen > existing.lastSeen) {
byName.set(key, peer);
}
}