T-HUB/packages/connector-reddit/src/index.ts
glass beadian aa5e262f3a Tactical Media Broadcast Hub — initial release
Self-hosted multi-platform broadcast newsroom for activist collectives:
publishing pipeline with group approval and DST-proof scheduling, TweetDeck-style
column wall with hostile-content-safe rendering, live control room with Auto
Director, bridged chat governance with polls and quorum, distributed sandboxed
transcode workers, encrypted content-addressed archive, monthly analytics.

MIT licensed. See README.md and docs/.
2026-10-10 11:56:42 +00:00

560 lines
18 KiB
TypeScript

/**
* Reddit connector — OAuth API access for posts, community feeds, comments,
* crossposts, inbox messages, and deletion.
*
* Reddit application registration happens at https://www.reddit.com/prefs/apps.
* The application exchanges authorization codes (and later refresh tokens) at
* https://www.reddit.com/api/v1/access_token using HTTP Basic client auth. The
* resulting access and refresh tokens are supplied as secret connection fields.
*/
import { randomUUID } from "node:crypto";
import type {
Comment,
ConnectorCapabilities,
ConnectorConnection,
DirectMessage,
FeedItem,
NormalizedIdentity,
NormalizedPost,
Page,
Publication,
} from "@tactical-media-broadcast-hub/core";
import {
assertValidManifest,
AuthError,
CapabilityNotSupportedError,
ConnectorError,
RateLimitedHttpClient,
type ConnectorManifest,
type PlatformConnector,
} from "@tactical-media-broadcast-hub/connector-sdk";
const PROVIDER = "reddit";
const API_BASE = "https://oauth.reddit.com";
const TITLE_LIMIT = 300;
const DEFAULT_FEED_LIMIT = 25;
export const REDDIT_RATE_POLICY = { requestsPerSecond: 0.8 } as const;
const CAPABILITIES: ConnectorCapabilities = {
publishText: true,
publishImages: false,
feed: true,
comments: true,
replies: true,
amplifyRepost: false,
amplifyQuote: false,
amplifyShare: true,
directMessages: true,
deletePublication: true,
webhooks: false,
nativeScheduling: false,
};
export interface RedditConnectorOptions {
/** Fetch override used by contract tests and embedding applications. */
fetchImpl?: typeof fetch;
}
export interface RedditAmplifyRequest {
kind: "repost" | "quote" | "share";
targetRemoteId: string;
commentary?: string;
quoteAssetIds?: string[];
/** Destination community for the crosspost. */
providerMetadata?: Record<string, unknown>;
}
export interface RedditConnector extends Omit<PlatformConnector, "amplify"> {
amplify(
connection: ConnectorConnection,
request: RedditAmplifyRequest,
): Promise<Publication>;
}
interface RedditMe {
id?: string;
name: string;
icon_img?: string;
}
interface RedditSubmitData {
id?: string;
name?: string;
url?: string;
things?: { data: RedditSubmitData }[];
}
interface RedditJsonResponse {
json: {
errors?: unknown[];
data?: RedditSubmitData;
};
}
interface RedditListing<T> {
data: {
after: string | null;
children: { kind: string; data: T }[];
};
}
interface RedditPost {
id: string;
name?: string;
title: string;
author: string;
score: number;
num_comments: number;
permalink: string;
selftext?: string;
created_utc: number;
url?: string;
}
interface RedditComment {
id: string;
name?: string;
parent_id?: string;
author: string;
body: string;
created_utc: number;
}
interface RedditMessage {
id: string;
name?: string;
author?: string;
dest?: string;
body: string;
created_utc: number;
first_message_name?: string;
parent_id?: string;
}
interface SubmitResult {
id: string;
url?: string;
}
function stringMetadata(metadata: Record<string, unknown>, key: string): string | undefined {
const value = metadata[key];
return typeof value === "string" && value.trim().length > 0 ? value.trim() : undefined;
}
function stripPrefix(value: string, prefix: string): string {
return value.startsWith(prefix) ? value.slice(prefix.length) : value;
}
function subredditName(value: string): string {
const normalized = value.trim().replace(/^\/?r\//i, "");
if (!/^[A-Za-z0-9_]{2,21}$/.test(normalized)) {
throw new ConnectorError(`${PROVIDER}: invalid subreddit "${value}"`, PROVIDER);
}
return normalized;
}
function fullname(remoteId: string, defaultKind: "t1" | "t3"): string {
if (/^t[13]_[A-Za-z0-9]+$/.test(remoteId)) return remoteId;
const id = remoteId.trim();
if (!/^[A-Za-z0-9]+$/.test(id)) {
throw new ConnectorError(`${PROVIDER}: invalid remote id "${remoteId}"`, PROVIDER);
}
return `${defaultKind}_${id}`;
}
function formPost(values: Record<string, string>): RequestInit {
return {
method: "POST",
headers: { "content-type": "application/x-www-form-urlencoded" },
body: new URLSearchParams(values),
};
}
function pageOf<T>(items: T[], nextCursor: string | null | undefined): Page<T> {
return nextCursor === null || nextCursor === undefined
? { items }
: { items, nextCursor };
}
function normalizeAvatar(url: string | undefined): string | undefined {
return url?.replace(/&amp;/g, "&");
}
function postText(post: RedditPost): string {
const body = post.selftext?.trim();
return body === undefined || body.length === 0 ? post.title : `${post.title}\n\n${body}`;
}
function titleAndBody(post: NormalizedPost): { title: string; text: string } {
const explicitTitle = stringMetadata(post.providerMetadata, "title");
if (explicitTitle !== undefined) return { title: explicitTitle, text: post.text };
const normalized = post.text.trim();
const newline = normalized.indexOf("\n");
if (newline === -1) return { title: normalized, text: "" };
return {
title: normalized.slice(0, newline).trim(),
text: normalized.slice(newline + 1).trim(),
};
}
export function createRedditConnector(options: RedditConnectorOptions = {}): RedditConnector {
const client = new RateLimitedHttpClient({
requestsPerSecond: REDDIT_RATE_POLICY.requestsPerSecond,
fetchImpl: options.fetchImpl,
});
const manifest: ConnectorManifest = {
id: PROVIDER,
displayName: "Reddit",
capabilities: CAPABILITIES,
connectionFields: [
{
name: "accessToken",
label: "OAuth access token",
required: true,
secret: true,
},
{
name: "refreshToken",
label: "OAuth refresh token",
required: true,
secret: true,
},
{
name: "username",
label: "Reddit username",
required: true,
secret: false,
placeholder: "your_username",
},
],
oauth: {
kind: "token",
scopes: ["identity", "submit", "read", "privatemessages", "edit"],
},
docsUrl: "https://github.com/reddit-archive/reddit/wiki/OAuth2",
};
function accessToken(connection: ConnectorConnection): string {
const token = connection.fields["accessToken"];
if (token === undefined || token.trim().length === 0) {
throw new AuthError(
`${PROVIDER}: missing OAuth access token in the accessToken connection field`,
PROVIDER,
false,
);
}
return token;
}
async function send(
connection: ConnectorConnection,
path: string,
init: RequestInit = {},
): Promise<Response> {
const response = await client.request(PROVIDER, `${API_BASE}${path}`, {
...init,
headers: {
accept: "application/json",
authorization: `Bearer ${accessToken(connection)}`,
"user-agent": "AnarchoPoster/0.1 (Reddit connector)",
...init.headers,
},
});
if (!response.ok) {
const message = `${PROVIDER}: HTTP ${response.status} for ${new URL(response.url || `${API_BASE}${path}`).pathname}`;
if (response.status === 401 || response.status === 403) {
throw new AuthError(message, PROVIDER, false);
}
throw new ConnectorError(message, PROVIDER, { status: response.status });
}
return response;
}
async function api<T>(
connection: ConnectorConnection,
path: string,
init: RequestInit = {},
): Promise<T> {
const response = await send(connection, path, init);
try {
return (await response.json()) as T;
} catch (cause) {
throw new ConnectorError(`${PROVIDER}: non-JSON response for ${path}`, PROVIDER, cause);
}
}
async function submit(
connection: ConnectorConnection,
values: Record<string, string>,
): Promise<SubmitResult> {
const response = await api<RedditJsonResponse>(
connection,
"/api/submit",
formPost({ api_type: "json", resubmit: "true", ...values }),
);
const errors = response.json.errors ?? [];
if (errors.length > 0) {
throw new ConnectorError(`${PROVIDER}: submit rejected: ${JSON.stringify(errors)}`, PROVIDER);
}
const data = response.json.data;
const id = data?.id ?? (data?.name === undefined ? undefined : stripPrefix(data.name, "t3_"));
if (id === undefined || id.length === 0) {
throw new ConnectorError(`${PROVIDER}: submit response did not include a post id`, PROVIDER);
}
return { id, url: data?.url };
}
function publication(
connection: ConnectorConnection,
result: SubmitResult,
draftId: string,
providerMetadata: Record<string, unknown>,
): Publication {
return {
id: randomUUID(),
workspaceId: connection.workspaceId,
draftId,
connectionId: connection.id,
remoteId: result.id,
remoteUrl: result.url,
publishedAt: new Date(),
providerMetadata,
};
}
return {
manifest: () => {
assertValidManifest(manifest);
return manifest;
},
async authorize() {
throw new ConnectorError(
`${PROVIDER}: register an app at reddit.com/prefs/apps, exchange the authorization code at /api/v1/access_token, then connect the access and refresh tokens`,
PROVIDER,
);
},
async getIdentity(connection: ConnectorConnection): Promise<NormalizedIdentity> {
const me = await api<RedditMe>(connection, "/api/v1/me");
return {
provider: PROVIDER,
remoteId: me.id ?? me.name,
handle: `u/${me.name}`,
displayName: me.name,
avatarUrl: normalizeAvatar(me.icon_img),
};
},
async publish(connection: ConnectorConnection, post: NormalizedPost): Promise<Publication> {
if (post.mediaAssetIds.length > 0) {
throw new CapabilityNotSupportedError(PROVIDER, "publishImages");
}
const srValue = stringMetadata(post.providerMetadata, "sr");
if (srValue === undefined) {
throw new ConnectorError(`${PROVIDER}: providerMetadata.sr is required`, PROVIDER);
}
const sr = subredditName(srValue);
const kind = stringMetadata(post.providerMetadata, "kind") ?? "self";
if (kind !== "self" && kind !== "link") {
throw new ConnectorError(`${PROVIDER}: providerMetadata.kind must be self or link`, PROVIDER);
}
const { title, text } = titleAndBody(post);
if (title.length === 0) {
throw new ConnectorError(`${PROVIDER}: post title is required`, PROVIDER);
}
if (title.length > TITLE_LIMIT) {
throw new ConnectorError(
`${PROVIDER}: title is ${title.length} characters, over the ${TITLE_LIMIT} limit`,
PROVIDER,
);
}
const values: Record<string, string> = { sr, kind, title };
if (kind === "self") {
values["text"] = text;
} else {
const url = stringMetadata(post.providerMetadata, "url");
if (url === undefined) {
throw new ConnectorError(`${PROVIDER}: providerMetadata.url is required for link posts`, PROVIDER);
}
values["url"] = url;
}
const result = await submit(connection, values);
return publication(connection, result, "direct", {
...post.providerMetadata,
sr,
kind,
});
},
async amplify(
connection: ConnectorConnection,
request: RedditAmplifyRequest,
): Promise<Publication> {
if (request.kind !== "share") {
throw new CapabilityNotSupportedError(PROVIDER, `amplify:${request.kind}`);
}
const srValue = request.providerMetadata === undefined
? undefined
: stringMetadata(request.providerMetadata, "sr");
if (srValue === undefined) {
throw new ConnectorError(
`${PROVIDER}: providerMetadata.sr is required for a crosspost`,
PROVIDER,
);
}
const sr = subredditName(srValue);
const originalId = stripPrefix(request.targetRemoteId, "t3_");
// Manipulation-policy boundary: amplification creates a user-requested
// crosspost. It never calls /api/vote or performs automated voting;
// voting belongs only to an explicit, single user click in a manual UI.
const result = await submit(connection, {
sr,
kind: "crosspost",
crosspost_fullname: fullname(originalId, "t3"),
...(request.commentary === undefined || request.commentary.length === 0
? {}
: { title: request.commentary.slice(0, TITLE_LIMIT) }),
});
return publication(connection, result, "amplify:share", {
sr,
kind: "crosspost",
amplifies: originalId,
});
},
async getFeed(
connection: ConnectorConnection,
request: { kind: "home" | "mentions" | "search"; query?: string; cursor?: string; limit: number },
): Promise<Page<FeedItem>> {
if (request.kind === "mentions") {
throw new CapabilityNotSupportedError(PROVIDER, "feed:mentions");
}
const srValue = request.query ?? connection.fields["subreddit"];
if (srValue === undefined || srValue.trim().length === 0) {
throw new ConnectorError(`${PROVIDER}: feed query must name a subreddit`, PROVIDER);
}
const sr = subredditName(srValue);
const limit = Math.max(1, Math.min(DEFAULT_FEED_LIMIT, request.limit));
const params = new URLSearchParams({ limit: String(limit) });
if (request.cursor !== undefined) params.set("after", request.cursor);
const listing = await api<RedditListing<RedditPost>>(
connection,
`/r/${encodeURIComponent(sr)}/hot?${params}`,
);
const items = listing.data.children.map<FeedItem>(({ data }) => ({
id: data.name ?? `t3_${data.id}`,
provider: PROVIDER,
remoteId: data.id,
authorHandle: `u/${data.author}`,
text: postText(data),
mediaUrls: (data.selftext ?? "").trim().length === 0 && data.url !== undefined ? [data.url] : [],
createdAt: new Date(data.created_utc * 1000),
metrics: { score: data.score, comments: data.num_comments },
}));
return pageOf(items, listing.data.after);
},
async getComments(
connection: ConnectorConnection,
request: { onRemoteId: string; cursor?: string; limit: number },
): Promise<Page<Comment>> {
const id = stripPrefix(request.onRemoteId, "t3_");
const limit = Math.max(1, Math.min(100, request.limit));
const params = new URLSearchParams({ limit: String(limit) });
if (request.cursor !== undefined) params.set("after", request.cursor);
const listings = await api<RedditListing<RedditPost | RedditComment>[]>(
connection,
`/comments/${encodeURIComponent(id)}.json?${params}`,
);
const comments = listings[1];
if (comments === undefined) return { items: [] };
const items = comments.data.children.flatMap<Comment>(({ kind, data }) => {
if (kind !== "t1" || !("body" in data)) return [];
return [{
id: data.name ?? `t1_${data.id}`,
provider: PROVIDER,
remoteId: data.id,
onRemoteId: data.parent_id ?? fullname(id, "t3"),
authorHandle: `u/${data.author}`,
text: data.body,
createdAt: new Date(data.created_utc * 1000),
}];
});
return pageOf(items, comments.data.after);
},
async reply(
connection: ConnectorConnection,
request: { onRemoteId: string; text: string },
): Promise<Publication> {
const response = await api<RedditJsonResponse>(
connection,
"/api/comment",
formPost({
api_type: "json",
thing_id: fullname(request.onRemoteId, "t3"),
text: request.text,
}),
);
const errors = response.json.errors ?? [];
if (errors.length > 0) {
throw new ConnectorError(`${PROVIDER}: comment rejected: ${JSON.stringify(errors)}`, PROVIDER);
}
const responseData = response.json.data;
const data = responseData?.things?.[0]?.data ?? responseData;
const id = data?.id ?? (data?.name === undefined ? undefined : stripPrefix(data.name, "t1_"));
if (id === undefined || id.length === 0) {
throw new ConnectorError(`${PROVIDER}: comment response did not include an id`, PROVIDER);
}
return publication(connection, { id, url: data?.url }, "reply", {
onRemoteId: request.onRemoteId,
});
},
async getMessages(
connection: ConnectorConnection,
request: { conversationRemoteId?: string; cursor?: string; limit: number },
): Promise<Page<DirectMessage>> {
const limit = Math.max(1, Math.min(100, request.limit));
const params = new URLSearchParams({ limit: String(limit) });
if (request.cursor !== undefined) params.set("after", request.cursor);
const listing = await api<RedditListing<RedditMessage>>(
connection,
`/api/v1/message/inbox?${params}`,
);
const messages = listing.data.children
.map(({ data }) => data)
.filter((message) => {
if (request.conversationRemoteId === undefined) return true;
const conversationId = message.first_message_name ?? message.parent_id ?? message.name ?? message.id;
return conversationId === request.conversationRemoteId;
})
.map<DirectMessage>((message) => ({
id: message.name ?? `t4_${message.id}`,
provider: PROVIDER,
conversationRemoteId: message.first_message_name ?? message.parent_id ?? message.name ?? message.id,
fromHandle: message.author === undefined ? "u/[deleted]" : `u/${message.author}`,
toHandle: `u/${message.dest ?? connection.fields["username"] ?? "unknown"}`,
text: message.body,
sentAt: new Date(message.created_utc * 1000),
}));
return pageOf(messages, listing.data.after);
},
async deletePublication(connection: ConnectorConnection, remoteId: string): Promise<void> {
// Successful /api/del replies may have an empty body, so deletion only
// relies on the HTTP status rather than attempting to decode JSON.
await send(
connection,
"/api/del",
formPost({ id: fullname(remoteId, "t3") }),
);
},
};
}