/**
 * Bluesky connector using the AT Protocol XRPC API.
 *
 * AT record operations need both a URI and CID. the connector
 * methods expose one remoteId, so this connector encodes record references as
 * `<at-uri>|<cid>`. The individual values are also retained in publication
 * metadata. Delete operations only need the URI and accept either form.
 */
import { randomUUID } from "node:crypto";
import type {
  Comment,
  ConnectorCapabilities,
  ConnectorConnection,
  CredentialRef,
  FeedItem,
  MediaAsset,
  NormalizedIdentity,
  NormalizedPost,
  Page,
  Publication,
  RemoteMedia,
} from "@tactical-media-broadcast-hub/core";
import {
  assertValidManifest,
  AuthError,
  CapabilityNotSupportedError,
  ConnectorError,
  InMemoryCredentials,
  pageOf,
  RateLimitedHttpClient,
  type AuthorizationRequest,
  type AuthorizationResult,
  type ConnectorManifest,
  type CredentialAccessor,
  type PlatformConnector,
} from "@tactical-media-broadcast-hub/connector-sdk";

const PROVIDER = "bluesky";
const DEFAULT_SERVICE = "https://bsky.social";
const POST_COLLECTION = "app.bsky.feed.post";
const REPOST_COLLECTION = "app.bsky.feed.repost";
const MAX_IMAGES = 4;

const SECRET = {
  accessToken: "access_token",
  refreshToken: "refresh_token",
  did: "did",
} as const;

const CAPABILITIES: ConnectorCapabilities = {
  publishText: true,
  publishImages: true,
  feed: true,
  mentions: true,
  replies: true,
  comments: true,
  amplifyRepost: true,
  amplifyQuote: true,
  deletePublication: true,
  directMessages: false,
  webhooks: false,
};

export interface BlueskyConnectorOptions {
  /** fetch override for tests and host applications. */
  fetchImpl?: typeof fetch;
  /** The documented default is one request per second. */
  requestsPerSecond?: number;
  /** Sleep override for deterministic tests. */
  sleep?: (ms: number) => Promise<void>;
  /** Secret store. Default: in-memory for development and contract tests. */
  credentials?: CredentialAccessor;
  /** Resolve local media ids used by NormalizedPost. */
  loadAsset?: (assetId: string) => Promise<MediaAsset>;
  /** Read media bytes for com.atproto.repo.uploadBlob. */
  openAsset?: (media: MediaAsset) => Promise<{ bytes: Uint8Array; filename: string }>;
}

export interface BlueskyAuthorizationRequest extends AuthorizationRequest {
  fields?: Record<string, string>;
  providerMetadata?: Record<string, unknown>;
}

export interface BlueskyConnector
  extends Omit<
    PlatformConnector,
    | "authorize"
    | "publish"
    | "deletePublication"
    | "uploadMedia"
    | "getFeed"
    | "getComments"
    | "reply"
    | "amplify"
    | "unamplify"
  > {
  authorize(input: BlueskyAuthorizationRequest): Promise<AuthorizationResult>;
  publish: NonNullable<PlatformConnector["publish"]>;
  deletePublication: NonNullable<PlatformConnector["deletePublication"]>;
  uploadMedia: NonNullable<PlatformConnector["uploadMedia"]>;
  getFeed: NonNullable<PlatformConnector["getFeed"]>;
  getComments: NonNullable<PlatformConnector["getComments"]>;
  reply: NonNullable<PlatformConnector["reply"]>;
  amplify: NonNullable<PlatformConnector["amplify"]>;
  unamplify: NonNullable<PlatformConnector["unamplify"]>;
}

interface SessionResponse {
  accessJwt: string;
  refreshJwt: string;
  did: string;
}

interface ProfileResponse {
  did: string;
  handle: string;
  displayName?: string;
  avatar?: string;
}

interface RecordReference {
  uri: string;
  cid: string;
}

interface StrongRef {
  uri: string;
  cid: string;
}

interface BlobReference {
  $type?: string;
  ref: { $link: string } | string;
  mimeType: string;
  size: number;
}

interface UploadBlobResponse {
  blob: BlobReference;
}

interface CreateRecordResponse extends RecordReference {
  validationStatus?: string;
}

interface PostRecord {
  text: string;
  createdAt: string;
  reply?: { root: StrongRef; parent: StrongRef };
}

interface PostView {
  uri: string;
  cid: string;
  author: {
    did: string;
    handle: string;
    displayName?: string;
    avatar?: string;
  };
  record: PostRecord;
  embed?: unknown;
  likeCount?: number;
  repostCount?: number;
  replyCount?: number;
}

interface FeedViewPost {
  post: PostView;
}

interface TimelineResponse {
  feed: FeedViewPost[];
  cursor?: string;
}

interface SearchResponse {
  posts: PostView[];
  cursor?: string;
}

interface NotificationView {
  uri: string;
  cid: string;
  author: PostView["author"];
  record: PostRecord;
  reason: string;
  indexedAt: string;
}

interface NotificationsResponse {
  notifications: NotificationView[];
  cursor?: string;
}

interface ThreadViewPost {
  post: PostView;
  replies?: ThreadViewPost[];
}

interface PostThreadResponse {
  thread: ThreadViewPost;
}

interface ParsedAtUri {
  repo: string;
  collection: string;
  rkey: string;
}

function isObject(value: unknown): value is Record<string, unknown> {
  return typeof value === "object" && value !== null;
}

function stringValue(value: unknown): string | undefined {
  return typeof value === "string" && value.length > 0 ? value : undefined;
}

function normalizeService(value: string | undefined): string {
  const raw = value === undefined || value.trim() === "" ? DEFAULT_SERVICE : value.trim();
  let url: URL;
  try {
    url = new URL(raw);
  } catch (cause) {
    throw new ConnectorError(`${PROVIDER}: invalid service URL`, PROVIDER, cause);
  }
  if (url.protocol !== "https:" && url.protocol !== "http:") {
    throw new ConnectorError(`${PROVIDER}: service must use http or https`, PROVIDER);
  }
  return url.toString().replace(/\/$/, "");
}

function jsonPost(body: Record<string, unknown>): RequestInit {
  return {
    method: "POST",
    headers: { "content-type": "application/json" },
    body: JSON.stringify(body),
  };
}

function encodeReference(reference: RecordReference): string {
  return `${reference.uri}|${reference.cid}`;
}

function parseAtUri(uri: string): ParsedAtUri {
  const match = /^at:\/\/([^/]+)\/([^/]+)\/([^/?#]+)$/.exec(uri);
  if (match === null) {
    throw new ConnectorError(`${PROVIDER}: malformed AT record URI "${uri}"`, PROVIDER);
  }
  const repo = match[1];
  const collection = match[2];
  const rkey = match[3];
  if (repo === undefined || collection === undefined || rkey === undefined) {
    throw new ConnectorError(`${PROVIDER}: incomplete AT record URI "${uri}"`, PROVIDER);
  }
  return { repo, collection, rkey };
}

function imageUrls(embed: unknown): string[] {
  if (!isObject(embed) || !Array.isArray(embed.images)) return [];
  return embed.images.flatMap((image) => {
    if (!isObject(image)) return [];
    const fullsize = stringValue(image.fullsize);
    const thumb = stringValue(image.thumb);
    return fullsize !== undefined ? [fullsize] : thumb !== undefined ? [thumb] : [];
  });
}

function blobCid(blob: BlobReference): string {
  if (typeof blob.ref === "string") return blob.ref;
  return blob.ref.$link;
}

export function createBlueskyConnector(options: BlueskyConnectorOptions = {}): BlueskyConnector {
  const client = new RateLimitedHttpClient({
    requestsPerSecond: options.requestsPerSecond ?? 1,
    fetchImpl: options.fetchImpl,
    sleep: options.sleep,
  });
  const credentials = options.credentials ?? new InMemoryCredentials();
  const references = new Map<string, RecordReference>();
  const replyRoots = new Map<string, StrongRef>();
  const blobReferences = new Map<string, BlobReference>();
  const handles = new Map<string, string>();

  const manifest: ConnectorManifest = {
    id: PROVIDER,
    displayName: "Bluesky",
    capabilities: CAPABILITIES,
    connectionFields: [
      {
        name: "identifier",
        label: "Handle or email",
        required: true,
        secret: false,
        placeholder: "you.bsky.social",
      },
      {
        // The SDK contract forbids "password" in field names so generic UIs
        // and logs do not mistake this for an account password.
        name: "appKey",
        label: "App password (never your account password)",
        required: true,
        secret: true,
        placeholder: "xxxx-xxxx-xxxx-xxxx",
      },
      {
        name: "service",
        label: "Personal data server",
        required: false,
        secret: false,
        placeholder: DEFAULT_SERVICE,
      },
    ],
    oauth: { kind: "token" },
    docsUrl: "https://docs.bsky.app/docs/category/http-reference",
  };

  function serviceOf(connection: ConnectorConnection): string {
    return normalizeService(connection.fields.service);
  }

  async function send(url: string, init: RequestInit): Promise<Response> {
    const response = await client.request(PROVIDER, url, {
      ...init,
      headers: { accept: "application/json", ...init.headers },
    });
    if (response.ok) return response;
    let detail = "";
    try {
      const body = (await response.json()) as { error?: string; message?: string };
      detail = body.message ?? body.error ?? "";
    } catch {
      detail = "";
    }
    const message = `${PROVIDER}: HTTP ${response.status} ${new URL(url).pathname}${detail ? `: ${detail}` : ""}`;
    if (response.status === 401 || response.status === 403) {
      throw new AuthError(message, PROVIDER, false);
    }
    throw new ConnectorError(message, PROVIDER, { status: response.status });
  }

  async function publicJson<T>(service: string, path: string, init: RequestInit): Promise<T> {
    const response = await send(`${service}/xrpc/${path}`, init);
    return (await response.json()) as T;
  }

  async function api<T>(
    connection: ConnectorConnection,
    path: string,
    init: RequestInit = {},
  ): Promise<T> {
    let token: string;
    try {
      token = await credentials.getAccessToken(connection);
    } catch {
      throw new AuthError(`${PROVIDER}: no access token stored for ${connection.id}`, PROVIDER, false);
    }
    const response = await send(`${serviceOf(connection)}/xrpc/${path}`, {
      ...init,
      headers: { authorization: `Bearer ${token}`, ...init.headers },
    });
    const text = await response.text();
    return (text.length === 0 ? {} : JSON.parse(text)) as T;
  }

  async function didOf(connection: ConnectorConnection): Promise<string> {
    if (connection.channelId.startsWith("did:")) return connection.channelId;
    try {
      return await credentials.getSecret(connection, SECRET.did);
    } catch {
      throw new AuthError(`${PROVIDER}: no DID stored for ${connection.id}`, PROVIDER, false);
    }
  }

  function rememberReference(reference: RecordReference, root?: StrongRef): string {
    const encoded = encodeReference(reference);
    references.set(encoded, reference);
    references.set(reference.uri, reference);
    if (root !== undefined) {
      replyRoots.set(encoded, root);
      replyRoots.set(reference.uri, root);
    }
    return encoded;
  }

  function requireReference(remoteId: string): RecordReference {
    const remembered = references.get(remoteId);
    if (remembered !== undefined) return remembered;
    const divider = remoteId.lastIndexOf("|");
    if (divider > 0 && divider < remoteId.length - 1) {
      const reference = { uri: remoteId.slice(0, divider), cid: remoteId.slice(divider + 1) };
      parseAtUri(reference.uri);
      references.set(remoteId, reference);
      references.set(reference.uri, reference);
      return reference;
    }
    throw new ConnectorError(
      `${PROVIDER}: record reference must include both URI and CID (expected "at://…|cid")`,
      PROVIDER,
    );
  }

  function uriFromRemoteId(remoteId: string): string {
    const remembered = references.get(remoteId);
    if (remembered !== undefined) return remembered.uri;
    const divider = remoteId.lastIndexOf("|");
    const uri = divider > 0 ? remoteId.slice(0, divider) : remoteId;
    parseAtUri(uri);
    return uri;
  }

  async function createRecord(
    connection: ConnectorConnection,
    collection: string,
    record: Record<string, unknown>,
  ): Promise<CreateRecordResponse> {
    return api<CreateRecordResponse>(
      connection,
      "com.atproto.repo.createRecord",
      jsonPost({ repo: await didOf(connection), collection, record }),
    );
  }

  function remoteUrl(connection: ConnectorConnection, uri: string): string | undefined {
    if (!uri.includes(`/${POST_COLLECTION}/`)) return undefined;
    const handle = handles.get(connection.id) ?? connection.fields.identifier;
    if (handle === undefined || handle.length === 0) return undefined;
    return `https://bsky.app/profile/${encodeURIComponent(handle)}/post/${encodeURIComponent(parseAtUri(uri).rkey)}`;
  }

  function toPublication(
    connection: ConnectorConnection,
    reference: CreateRecordResponse,
    draftId: string,
    publishedAt: Date,
    providerMetadata: Record<string, unknown> = {},
  ): Publication {
    const remoteId = rememberReference(reference);
    const publication: Publication = {
      id: randomUUID(),
      workspaceId: connection.workspaceId,
      draftId,
      connectionId: connection.id,
      remoteId,
      publishedAt,
      providerMetadata: {
        uri: reference.uri,
        cid: reference.cid,
        collection: parseAtUri(reference.uri).collection,
        ...providerMetadata,
      },
    };
    const url = remoteUrl(connection, reference.uri);
    if (url !== undefined) publication.remoteUrl = url;
    return publication;
  }

  function toFeedItem(post: PostView): FeedItem {
    const root = post.record.reply?.root;
    const remoteId = rememberReference(post, root);
    return {
      id: post.uri,
      provider: PROVIDER,
      remoteId,
      authorHandle: post.author.handle,
      text: post.record.text,
      mediaUrls: imageUrls(post.embed),
      createdAt: new Date(post.record.createdAt),
      metrics: {
        likes: post.likeCount ?? 0,
        reposts: post.repostCount ?? 0,
        replies: post.replyCount ?? 0,
      },
    };
  }

  async function uploadAsset(connection: ConnectorConnection, media: MediaAsset): Promise<RemoteMedia> {
    if (!media.mimeType.startsWith("image/")) {
      throw new ConnectorError(`${PROVIDER}: uploadMedia accepts images only`, PROVIDER);
    }
    if (options.openAsset === undefined) {
      throw new ConnectorError(`${PROVIDER}: media upload requires the openAsset option`, PROVIDER);
    }
    const { bytes } = await options.openAsset(media);
    const response = await api<UploadBlobResponse>(
      connection,
      "com.atproto.repo.uploadBlob",
      {
        method: "POST",
        headers: { "content-type": media.mimeType },
        body: new Blob([new Uint8Array(bytes)], { type: media.mimeType }),
      },
    );
    const cid = blobCid(response.blob);
    blobReferences.set(cid, response.blob);
    const remote: RemoteMedia = { remoteId: cid, mimeType: response.blob.mimeType };
    if (media.width !== undefined) remote.width = media.width;
    if (media.height !== undefined) remote.height = media.height;
    return remote;
  }

  async function imagesFor(
    connection: ConnectorConnection,
    post: NormalizedPost,
  ): Promise<{ alt: string; image: BlobReference }[]> {
    if (post.mediaAssetIds.length > MAX_IMAGES) {
      throw new ConnectorError(`${PROVIDER}: posts support at most ${MAX_IMAGES} images`, PROVIDER);
    }
    const altText = isObject(post.providerMetadata.altText) ? post.providerMetadata.altText : {};
    const images: { alt: string; image: BlobReference }[] = [];
    for (const assetId of post.mediaAssetIds) {
      let blob = blobReferences.get(assetId);
      if (blob === undefined) {
        if (options.loadAsset === undefined) {
          throw new ConnectorError(
            `${PROVIDER}: publishing local media requires the loadAsset option`,
            PROVIDER,
          );
        }
        const remote = await uploadAsset(connection, await options.loadAsset(assetId));
        blob = blobReferences.get(remote.remoteId);
      }
      if (blob === undefined) {
        throw new ConnectorError(`${PROVIDER}: upload returned no blob reference`, PROVIDER);
      }
      images.push({ alt: stringValue(altText[assetId]) ?? "", image: blob });
    }
    return images;
  }

  async function deleteRecord(connection: ConnectorConnection, remoteId: string): Promise<void> {
    const parsed = parseAtUri(uriFromRemoteId(remoteId));
    await api<Record<string, never>>(
      connection,
      "com.atproto.repo.deleteRecord",
      jsonPost({ repo: await didOf(connection), collection: parsed.collection, rkey: parsed.rkey }),
    );
  }

  async function identityFor(connection: ConnectorConnection): Promise<NormalizedIdentity> {
    const did = await didOf(connection);
    const profile = await api<ProfileResponse>(
      connection,
      `app.bsky.actor.getProfile?actor=${encodeURIComponent(did)}`,
    );
    handles.set(connection.id, profile.handle);
    const identity: NormalizedIdentity = {
      provider: PROVIDER,
      remoteId: profile.did,
      handle: profile.handle,
    };
    if (profile.displayName !== undefined && profile.displayName !== "") {
      identity.displayName = profile.displayName;
    }
    if (profile.avatar !== undefined && profile.avatar !== "") identity.avatarUrl = profile.avatar;
    return identity;
  }

  return {
    manifest: () => {
      assertValidManifest(manifest);
      return manifest;
    },

    async authorize(input: BlueskyAuthorizationRequest): Promise<AuthorizationResult> {
      const fields = input.fields ?? {};
      const identifier = fields.identifier?.trim();
      // appPassword remains accepted for callers already using AT's common
      // terminology; appKey is the SDK-safe manifest field name.
      const appPassword = (fields.appKey ?? fields.appPassword)?.trim();
      if (identifier === undefined || identifier === "") {
        throw new AuthError(`${PROVIDER}: authorize needs fields.identifier`, PROVIDER, false);
      }
      if (appPassword === undefined || appPassword === "") {
        throw new AuthError(
          `${PROVIDER}: authorize needs fields.appKey (an app password from Bluesky Settings, never the account password)`,
          PROVIDER,
          false,
        );
      }
      const service = normalizeService(fields.service);
      const session = await publicJson<SessionResponse>(
        service,
        "com.atproto.server.createSession",
        jsonPost({ identifier, password: appPassword }),
      );
      const metadata = input.providerMetadata ?? {};
      const connectionId = stringValue(metadata.connectionId) ?? `${PROVIDER}:${input.workspaceId}:${session.did}`;
      const credentialId: CredentialRef = { kind: "credential-ref", id: connectionId };
      const connection: ConnectorConnection = {
        id: connectionId,
        workspaceId: input.workspaceId,
        provider: PROVIDER,
        channelId: session.did,
        credentialId,
        fields: { identifier, service },
        createdAt: new Date(),
      };
      await credentials.put(connection, SECRET.accessToken, session.accessJwt);
      await credentials.put(connection, SECRET.refreshToken, session.refreshJwt);
      await credentials.put(connection, SECRET.did, session.did);
      const identity = await identityFor(connection);
      return { connection, identity };
    },

    async getIdentity(connection: ConnectorConnection): Promise<NormalizedIdentity> {
      return identityFor(connection);
    },

    async publish(connection: ConnectorConnection, post: NormalizedPost): Promise<Publication> {
      const createdAt = new Date();
      const record: Record<string, unknown> = { text: post.text, createdAt: createdAt.toISOString() };
      const images = await imagesFor(connection, post);
      if (images.length > 0) {
        record.embed = { $type: "app.bsky.embed.images", images };
      }
      const response = await createRecord(connection, POST_COLLECTION, record);
      const draftId = stringValue(post.providerMetadata.draftId) ?? "direct";
      return toPublication(connection, response, draftId, createdAt);
    },

    async uploadMedia(connection: ConnectorConnection, media: MediaAsset): Promise<RemoteMedia> {
      return uploadAsset(connection, media);
    },

    async getFeed(
      connection: ConnectorConnection,
      request: { kind: "home" | "mentions" | "search"; query?: string; cursor?: string; limit: number },
    ): Promise<Page<FeedItem>> {
      const limit = Math.max(1, Math.min(100, request.limit));
      const params = new URLSearchParams({ limit: String(limit) });
      if (request.cursor !== undefined) params.set("cursor", request.cursor);

      if (request.kind === "home") {
        const data = await api<TimelineResponse>(
          connection,
          `app.bsky.feed.getTimeline?${params.toString()}`,
        );
        return pageOf(data.feed.map((entry) => toFeedItem(entry.post)), data.cursor);
      }
      if (request.kind === "search") {
        const query = request.query?.trim();
        if (query === undefined || query === "") {
          throw new ConnectorError(`${PROVIDER}: search requires a query`, PROVIDER);
        }
        params.set("q", query);
        const data = await api<SearchResponse>(
          connection,
          `app.bsky.feed.searchPosts?${params.toString()}`,
        );
        return pageOf(data.posts.map(toFeedItem), data.cursor);
      }

      const data = await api<NotificationsResponse>(
        connection,
        `app.bsky.notification.listNotifications?${params.toString()}`,
      );
      const items = data.notifications
        .filter((notification) => notification.reason === "mention" || notification.reason === "reply")
        .map((notification) =>
          toFeedItem({
            ...notification,
            embed: undefined,
            likeCount: 0,
            repostCount: 0,
            replyCount: 0,
          }),
        );
      return pageOf(items, data.cursor);
    },

    async getComments(
      connection: ConnectorConnection,
      request: { onRemoteId: string; cursor?: string; limit: number },
    ): Promise<Page<Comment>> {
      const target = requireReference(request.onRemoteId);
      const params = new URLSearchParams({ uri: target.uri, depth: "1", parentHeight: "0" });
      const data = await api<PostThreadResponse>(
        connection,
        `app.bsky.feed.getPostThread?${params.toString()}`,
      );
      const replies = data.thread.replies ?? [];
      const offset = request.cursor === undefined ? 0 : Number(request.cursor);
      if (!Number.isInteger(offset) || offset < 0) {
        throw new ConnectorError(`${PROVIDER}: invalid comments cursor`, PROVIDER);
      }
      const slice = replies.slice(offset, offset + request.limit);
      const items = slice.map((reply): Comment => ({
        id: reply.post.uri,
        provider: PROVIDER,
        remoteId: rememberReference(reply.post, reply.post.record.reply?.root),
        onRemoteId: request.onRemoteId,
        authorHandle: reply.post.author.handle,
        text: reply.post.record.text,
        createdAt: new Date(reply.post.record.createdAt),
      }));
      const next = offset + slice.length;
      return pageOf(items, next < replies.length ? String(next) : undefined);
    },

    async reply(
      connection: ConnectorConnection,
      request: { onRemoteId: string; text: string },
    ): Promise<Publication> {
      const parent = requireReference(request.onRemoteId);
      const root = replyRoots.get(request.onRemoteId) ?? parent;
      const createdAt = new Date();
      const response = await createRecord(connection, POST_COLLECTION, {
        text: request.text,
        createdAt: createdAt.toISOString(),
        reply: { root, parent },
      });
      rememberReference(response, root);
      return toPublication(connection, response, "reply", createdAt, {
        reply: { root, parent },
      });
    },

    async amplify(
      connection: ConnectorConnection,
      request: {
        kind: "repost" | "quote" | "share";
        targetRemoteId: string;
        commentary?: string;
        quoteAssetIds?: string[];
      },
    ): Promise<Publication> {
      const target = requireReference(request.targetRemoteId);
      const createdAt = new Date();
      if (request.kind === "repost") {
        const response = await createRecord(connection, REPOST_COLLECTION, {
          subject: target,
          createdAt: createdAt.toISOString(),
        });
        return toPublication(connection, response, "amplify:repost", createdAt, {
          amplifies: target,
          kind: "repost",
        });
      }
      if (request.kind === "quote") {
        if ((request.quoteAssetIds?.length ?? 0) > 0) {
          throw new CapabilityNotSupportedError(PROVIDER, "quote posts with images");
        }
        const response = await createRecord(connection, POST_COLLECTION, {
          text: request.commentary ?? "",
          createdAt: createdAt.toISOString(),
          embed: { $type: "app.bsky.embed.record", record: target },
        });
        return toPublication(connection, response, "amplify:quote", createdAt, {
          amplifies: target,
          kind: "quote",
        });
      }
      throw new CapabilityNotSupportedError(PROVIDER, "amplifyShare");
    },

    async unamplify(connection: ConnectorConnection, remoteId: string): Promise<void> {
      await deleteRecord(connection, remoteId);
    },

    async deletePublication(connection: ConnectorConnection, remoteId: string): Promise<void> {
      await deleteRecord(connection, remoteId);
    },
  };
}