codefeed/ts/index.ts

import * as plugins from './plugins.js';

export class CodeFeed {
  private baseUrl: string;
  private token?: string;
  private lastRunTimestamp: string;
  // Raw changelog content for the current repository
  private changelogContent: string = '';
  // npm registry helper for published-on-npm checks
  private npmRegistry: plugins.smartnpm.NpmRegistry;
  // In-memory stateful cache of commits
  private enableCache: boolean = false;
  private cacheWindowMs?: number;
  private cache: plugins.interfaces.ICommitResult[] = [];
  // enable or disable npm publishedOnNpm checks (true by default)
  private enableNpmCheck: boolean = true;

  constructor(
    baseUrl: string,
    token?: string,
    lastRunTimestamp?: string,
    options?: {
      enableCache?: boolean;
      cacheWindowMs?: number;
      enableNpmCheck?: boolean;
    }
  ) {
    this.baseUrl = baseUrl;
    this.token = token;
    this.lastRunTimestamp =
      lastRunTimestamp ?? new Date(Date.now() - 7 * 24 * 60 * 60 * 1000).toISOString();
    // configure stateful caching
    this.enableCache = options?.enableCache ?? false;
    this.cacheWindowMs = options?.cacheWindowMs;
    this.enableNpmCheck = options?.enableNpmCheck ?? true;
    this.cache = [];
    // npm registry instance for version lookups
    this.npmRegistry = new plugins.smartnpm.NpmRegistry();
    console.log('CodeFeed initialized with last run timestamp:', this.lastRunTimestamp);
  }

  /**
   * Fetch all new commits (since lastRunTimestamp) across all orgs and repos.
   */
  public async fetchAllCommitsFromInstance(): Promise<plugins.interfaces.ICommitResult[]> {
    // Controlled concurrency with AsyncExecutionStack
    const stack = new plugins.lik.AsyncExecutionStack();
    stack.setNonExclusiveMaxConcurrency(5);
    // determine since timestamp for this run (stateful caching)
    let effectiveSince = this.lastRunTimestamp;
    if (this.enableCache && this.cache.length > 0) {
      // use newest timestamp in cache to fetch only tail
      effectiveSince = this.cache.reduce(
        (max, c) => (c.timestamp > max ? c.timestamp : max),
        effectiveSince
      );
    }

    // 1) get all organizations
    const orgs = await this.fetchAllOrganizations();

    // 2) fetch repos per org in parallel
    const repoLists = await Promise.all(
      orgs.map((org) =>
        stack.getNonExclusiveExecutionSlot(() => this.fetchRepositoriesForOrg(org))
      )
    );
    // flatten to [{ owner, name }]
    const allRepos = orgs.flatMap((org, i) =>
      repoLists[i].map((r) => ({ owner: org, name: r.name }))
    );

    // 3) probe latest commit per repo and fetch full list only if new commits exist
    const commitJobs = allRepos.map(({ owner, name }) =>
      stack.getNonExclusiveExecutionSlot(async () => {
        try {
          // 3a) Probe the most recent commit (limit=1)
          const probeResp = await this.fetchFunction(
            `/api/v1/repos/${owner}/${name}/commits?limit=1`,
            { headers: this.token ? { Authorization: `token ${this.token}` } : {} }
          );
          if (!probeResp.ok) {
            throw new Error(`Probe failed for ${owner}/${name}: ${probeResp.statusText}`);
          }
          const probeData: plugins.interfaces.ICommit[] = await probeResp.json();
          // If no commits or no new commits since last run, skip
          if (
            probeData.length === 0 ||
            new Date(probeData[0].commit.author.date).getTime() <=
              new Date(effectiveSince).getTime()
          ) {
            return { owner, name, commits: [] };
          }
          // 3b) Fetch commits since last run
          const commits = await this.fetchRecentCommitsForRepo(
            owner,
            name,
            effectiveSince
          );
          return { owner, name, commits };
        } catch (e: any) {
          console.error(`Failed to fetch commits for ${owner}/${name}:`, e.message);
          return { owner, name, commits: [] };
        }
      })
    );
    const commitResults = await Promise.all(commitJobs);

    // 4) build new commit entries with tagging, npm and changelog support
    const newResults: plugins.interfaces.ICommitResult[] = [];
    for (const { owner, name, commits } of commitResults) {
      // skip repos with no new commits
      if (commits.length === 0) {
        this.changelogContent = '';
        continue;
      }
      // load changelog for this repo
      await this.loadChangelogFromRepo(owner, name);
      // fetch tags for this repo
      let taggedShas: Set<string>;
      try {
        taggedShas = await this.fetchTags(owner, name);
      } catch (e: any) {
        console.error(`Failed to fetch tags for ${owner}/${name}:`, e.message);
        taggedShas = new Set<string>();
      }
      // fetch npm package info only if any new commits correspond to a tag
      const hasTaggedCommit = commits.some((c) => taggedShas.has(c.sha));
      let pkgInfo: { allVersions: Array<{ version: string }> } | null = null;
      if (hasTaggedCommit && this.enableNpmCheck) {
        try {
          pkgInfo = await this.npmRegistry.getPackageInfo(`@${owner}/${name}`);
        } catch (e: any) {
          console.error(`Failed to fetch package info for ${owner}/${name}:`, e.message);
          pkgInfo = null;
        }
      }
      // build commit entries
      for (const c of commits) {
        const versionCandidate = c.commit.message.replace(/\n/g, '').trim();
        const isTagged = taggedShas.has(c.sha);
        const publishedOnNpm = isTagged && pkgInfo
          ? pkgInfo.allVersions.some((v) => v.version === versionCandidate)
          : false;
        let changelogEntry: string | undefined;
        if (this.changelogContent) {
          changelogEntry = this.getChangelogForVersion(versionCandidate);
        }
        newResults.push({
          baseUrl: this.baseUrl,
          org: owner,
          repo: name,
          timestamp: c.commit.author.date,
          prettyAgoTime: plugins.smarttime.getMilliSecondsAsHumanReadableAgoTime(
            new Date(c.commit.author.date).getTime()
          ),
          hash: c.sha,
          commitMessage: c.commit.message,
          tagged: isTagged,
          publishedOnNpm,
          changelog: changelogEntry,
        });
      }
    }
    // if caching is enabled, merge into in-memory cache and return full cache
    if (this.enableCache) {
      const existingHashes = new Set(this.cache.map((c) => c.hash));
      const uniqueNew = newResults.filter((c) => !existingHashes.has(c.hash));
      this.cache.push(...uniqueNew);
      // trim commits older than window
      if (this.cacheWindowMs !== undefined) {
        const cutoff = Date.now() - this.cacheWindowMs;
        this.cache = this.cache.filter((c) => new Date(c.timestamp).getTime() >= cutoff);
      }
      // advance lastRunTimestamp to now
      this.lastRunTimestamp = new Date().toISOString();
      // sort descending by timestamp
      this.cache.sort((a, b) => b.timestamp.localeCompare(a.timestamp));
      return this.cache;
    }
    // otherwise, return only newly fetched commits
    return newResults;
  }

  /**
   * Load the changelog directly from the Gitea repository.
   */
  private async loadChangelogFromRepo(owner: string, repo: string): Promise<void> {
    const url = `/api/v1/repos/${owner}/${repo}/contents/changelog.md`;
    const headers: Record<string, string> = {};
    if (this.token) {
      headers['Authorization'] = `token ${this.token}`;
    }

    const response = await this.fetchFunction(url, { headers });
    if (!response.ok) {
      console.error(
        `Could not fetch CHANGELOG.md from ${owner}/${repo}: ${response.status} ${response.statusText}`
      );
      this.changelogContent = '';
      return;
    }

    const data = await response.json();
    if (!data.content) {
      console.warn(`No content field found in response for ${owner}/${repo}/changelog.md`);
      this.changelogContent = '';
      return;
    }

    // decode base64 content
    this.changelogContent = Buffer.from(data.content, 'base64').toString('utf8');
  }

  /**
   * Parse the changelog to find the entry for a given version.
   * The changelog format is assumed as:
   *
   * # Changelog
   *
   * ## <date> - <version> - <description>
   * <changes...>
   */
  private getChangelogForVersion(version: string): string | undefined {
    if (!this.changelogContent) {
      return undefined;
    }
    const lines = this.changelogContent.split('\n');
    const versionHeaderIndex = lines.findIndex((line) => line.includes(`- ${version} -`));
    if (versionHeaderIndex === -1) {
      return undefined;
    }

    const changelogLines: string[] = [];
    for (let i = versionHeaderIndex + 1; i < lines.length; i++) {
      const line = lines[i];
      // The next version header starts with `## `
      if (line.startsWith('## ')) {
        break;
      }
      changelogLines.push(line);
    }

    return changelogLines.join('\n').trim();
  }
  /**
   * Fetch all tags for a given repo and return the set of tagged commit SHAs
   */
  private async fetchTags(owner: string, repo: string): Promise<Set<string>> {
    const taggedShas = new Set<string>();
    let page = 1;
    while (true) {
      const url = `/api/v1/repos/${owner}/${repo}/tags?limit=50&page=${page}`;
      const resp = await this.fetchFunction(url, {
        headers: this.token ? { Authorization: `token ${this.token}` } : {},
      });
      if (!resp.ok) {
        console.error(`Failed to fetch tags for ${owner}/${repo}: ${resp.status} ${resp.statusText}`);
        return taggedShas;
      }
      const data: plugins.interfaces.ITag[] = await resp.json();
      if (data.length === 0) break;
      for (const t of data) {
        if (t.commit?.sha) taggedShas.add(t.commit.sha);
      }
      if (data.length < 50) break;
      page++;
    }
    return taggedShas;
  }

  private async fetchAllOrganizations(): Promise<string[]> {
    const resp = await this.fetchFunction('/api/v1/orgs', {
      headers: this.token ? { Authorization: `token ${this.token}` } : {},
    });
    if (!resp.ok) {
      throw new Error(`Failed to fetch organizations: ${resp.statusText}`);
    }
    const data: { username: string }[] = await resp.json();
    return data.map((o) => o.username);
  }

  private async fetchRepositoriesForOrg(org: string): Promise<plugins.interfaces.IRepository[]> {
    const resp = await this.fetchFunction(`/api/v1/orgs/${org}/repos?limit=50`, {
      headers: this.token ? { Authorization: `token ${this.token}` } : {},
    });
    if (!resp.ok) {
      throw new Error(`Failed to fetch repositories for ${org}: ${resp.statusText}`);
    }
    const data: plugins.interfaces.IRepository[] = await resp.json();
    return data;
  }

  private async fetchRecentCommitsForRepo(
    owner: string,
    repo: string,
    sinceTimestamp?: string
  ): Promise<plugins.interfaces.ICommit[]> {
    const since = sinceTimestamp ?? this.lastRunTimestamp;
    const resp = await this.fetchFunction(
      `/api/v1/repos/${owner}/${repo}/commits?since=${encodeURIComponent(
        since
      )}&limit=50`,
      { headers: this.token ? { Authorization: `token ${this.token}` } : {} }
    );
    if (!resp.ok) {
      throw new Error(`Failed to fetch commits for ${owner}/${repo}: ${resp.statusText}`);
    }
    const data: plugins.interfaces.ICommit[] = await resp.json();
    return data;
  }

  public async fetchFunction(
    urlArg: string,
    optionsArg: RequestInit = {}
  ): Promise<Response> {
    return fetch(`${this.baseUrl}${urlArg}`, optionsArg);
  }
}