feat: implement StepTimeline for retrieving authenticated user's timeline

Author: pirhoo

docs/guide/features.md

@@ -36,7 +36,7 @@ Trotsky provides comprehensive support for the Bluesky AT Protocol. Below is a l
  **StepStarterPack** | :white_check_mark: | Get a starter pack by its URI. | ```Trotsky.init(agent).starterPack("at://...").run()```
  **StepStarterPacks** | :white_check_mark: | Get a list of starter packs by their URIs. | ```Trotsky.init(agent).starterPacks([uri1, uri2]).each()```
  **StepStreamPosts** | :test_tube: | Stream posts from the firehose. | ```Trotsky.init(agent).streamPosts().each()```
- **StepTimeline** | :x: | Get the authenticated user's timeline. | ```Trotsky.init(agent).timeline().take(20).each()```
+ **StepTimeline** | :white_check_mark: | Get the authenticated user's timeline. | ```Trotsky.init(agent).timeline().take(20).each()```
 
 ## Planned Features
 

lib/core/StepTimeline.ts

@@ -0,0 +1,116 @@
+import type { AppBskyFeedGetTimeline, AppBskyFeedDefs, AtpAgent } from "@atproto/api"
+
+import { StepPosts, type StepPostsOutput } from "./StepPosts"
+
+/**
+ * Type representing the output of the timeline retrieved by {@link StepTimeline}.
+ * @public
+ */
+export type StepTimelineOutput = StepPostsOutput
+
+/**
+ * Type representing the query parameters for retrieving the timeline.
+ * @public
+ */
+export type StepTimelineQueryParams = AppBskyFeedGetTimeline.QueryParams
+
+/**
+ * Type representing the cursor for paginated queries.
+ * @public
+ */
+export type StepTimelineQueryParamsCursor = StepTimelineQueryParams["cursor"] | undefined
+
+/**
+ * Represents a step for retrieving the authenticated user's timeline using the Bluesky API.
+ * Supports paginated retrieval of posts from followed accounts.
+ *
+ * @typeParam P - Type of the parent step.
+ * @typeParam C - Type of the context object, defaulting to `null`.
+ * @typeParam O - Type of the output object, extending {@link StepTimelineOutput}.
+ *
+ * @example
+ * Get recent posts from your timeline:
+ * ```ts
+ * await Trotsky.init(agent)
+ *   .timeline()
+ *   .take(20)
+ *   .each()
+ *   .tap((step) => {
+ *     console.log(`@${step.context.author.handle}: ${step.context.record.text}`)
+ *   })
+ *   .run()
+ * ```
+ *
+ * @example
+ * Like posts from your timeline with specific criteria:
+ * ```ts
+ * await Trotsky.init(agent)
+ *   .timeline()
+ *   .take(50)
+ *   .each()
+ *   .when((step) => step?.context?.record?.text?.includes("#typescript"))
+ *   .like()
+ *   .wait(1000)
+ *   .run()
+ * ```
+ *
+ * @example
+ * Use a custom algorithm for timeline:
+ * ```ts
+ * await Trotsky.init(agent)
+ *   .timeline({ algorithm: "reverse-chronological" })
+ *   .take(10)
+ *   .each()
+ *   .run()
+ * ```
+ *
+ * @public
+ */
+export class StepTimeline<P, C = null, O extends StepTimelineOutput = StepTimelineOutput> extends StepPosts<P, C, O> {
+
+  /**
+   * Query parameters for the timeline request.
+   */
+  _queryParams: StepTimelineQueryParams
+
+  /**
+   * Initializes the StepTimeline instance with the given agent, parent, and optional query parameters.
+   *
+   * @param agent - The AT protocol agent used for API calls.
+   * @param parent - The parent step in the chain.
+   * @param queryParams - Optional query parameters for the timeline (e.g., algorithm).
+   */
+  constructor (agent: AtpAgent, parent: P, queryParams: StepTimelineQueryParams = {}) {
+    super(agent, parent)
+    this._queryParams = queryParams
+  }
+
+  /**
+   * Clones the current step and returns a new instance with the same parameters.
+   * @returns A new {@link StepTimeline} instance.
+   */
+  override clone () {
+    return super.clone(this._queryParams)
+  }
+
+  /**
+   * Applies pagination to retrieve timeline posts and sets the output.
+   * Fetches paginated results using the agent and appends them to the output.
+   */
+  async applyPagination () {
+    const feed = await this.paginate<AppBskyFeedDefs.FeedViewPost[], AppBskyFeedGetTimeline.Response>("feed", (cursor) => {
+      return this.agent.app.bsky.feed.getTimeline(this.queryParams(cursor))
+    })
+
+    this.output = feed.map((post: AppBskyFeedDefs.FeedViewPost) => post.post) as O
+  }
+
+  /**
+   * Generates query parameters for retrieving the timeline, including the optional cursor.
+   * @param cursor - The cursor for paginated queries.
+   * @returns The query parameters for retrieving the timeline.
+   */
+  queryParams (cursor: StepTimelineQueryParamsCursor): StepTimelineQueryParams {
+    return { ...this._queryParams, cursor }
+  }
+}

lib/core/Trotsky.ts

@@ -7,6 +7,7 @@ import type { StepPostsUris } from "./StepPosts"
 import type { StepCreatePostParams } from "./StepCreatePost"
 import type { StepWhenPredicate } from "./StepWhen"
 import type { StepTapInterceptor } from "./StepTap"
+import type { StepTimelineQueryParams } from "./StepTimeline"
 import type { Resolvable } from "./utils/resolvable"
 import type { StepStarterPackUri } from "./StepStarterPack"
 import type { StepStarterPacksUris } from "./StepStarterPacks"
@@ -31,7 +32,8 @@ import {
   StepActorsParam,
   StepPosts,
   StepSave,
-  StepSavePath
+  StepSavePath,
+  StepTimeline
 } from "../trotsky"
 
 /**
@@ -141,6 +143,15 @@ export class Trotsky extends StepBuilder  {
     return this.append(StepStreamPosts<this>) as T
   }
 
+  /**
+   * Adds a {@link StepTimeline} step.
+   * @param queryParams - Optional query parameters for the timeline.
+   * @returns The new {@link StepTimeline} instance.
+   */
+  timeline (queryParams: StepTimelineQueryParams = {}): StepTimeline<this> {
+    return this.append(StepTimeline<this>, queryParams)
+  }
+
   /**
    * Adds a {@link StepSave} step.
    * @param path - The path of the JSON file to save the output. If not provided, the file path will be created using the current step name.

lib/trotsky.ts

@@ -65,6 +65,7 @@ export * from "./core/StepActorsEntry"
 export * from "./core/StepPosts"
 export * from "./core/StepPostsEntry"
 export * from "./core/StepSearchPosts"
+export * from "./core/StepTimeline"
 
 // List of lists
 export * from "./core/StepLists"

tests/core/StepTimeline.test.ts

@@ -0,0 +1,218 @@
+import { afterAll, beforeAll, describe, expect, test } from "@jest/globals"
+import { TestNetwork, SeedClient, usersSeed } from "@atproto/dev-env"
+import { AtpAgent } from "@atproto/api"
+import { Trotsky, StepTimeline } from "../../lib/trotsky"
+
+describe("StepTimeline", () => {
+  let network: TestNetwork
+  let agent: AtpAgent
+  let sc: SeedClient
+
+  // accounts
+  let bob: { "did": string; "handle": string; "password": string }
+  let alice: { "did": string; "handle": string; "password": string }
+  let carol: { "did": string; "handle": string; "password": string }
+
+  beforeAll(async () => {
+    network = await TestNetwork.create({
+      "dbPostgresSchema": "trotsky_step_timeline"
+    })
+
+    agent = network.pds.getClient()
+    sc = network.getSeedClient()
+
+    // Seed users
+    await usersSeed(sc)
+    bob = sc.accounts[sc.dids.bob]
+    alice = sc.accounts[sc.dids.alice]
+    carol = sc.accounts[sc.dids.carol]
+
+    // Login as bob
+    await agent.login({ "identifier": bob.handle, "password": bob.password })
+
+    // Bob follows Alice and Carol
+    await agent.app.bsky.graph.follow.create(
+      { "repo": bob.did },
+      { "subject": alice.did, "createdAt": new Date().toISOString() }
+    )
+    await agent.app.bsky.graph.follow.create(
+      { "repo": bob.did },
+      { "subject": carol.did, "createdAt": new Date().toISOString() }
+    )
+
+    // Alice and Carol create posts
+    await sc.post(alice.did, "Alice's first post about TypeScript")
+    await sc.post(alice.did, "Alice's second post about JavaScript")
+    await sc.post(carol.did, "Carol's post about Python")
+    await sc.post(carol.did, "Carol's post about Rust")
+
+    await network.processAll()
+  }, 120e3)
+
+  afterAll(async () => {
+    await network.close()
+  })
+
+  test("should clone properly", () => {
+    const step = Trotsky.init(agent).timeline()
+    const cloned = step.clone()
+    expect(cloned).toBeInstanceOf(StepTimeline)
+  })
+
+  test("should get timeline posts", async () => {
+    const timeline = await Trotsky.init(agent)
+      .timeline()
+      .runHere()
+
+    expect(timeline).toBeInstanceOf(StepTimeline)
+    expect(timeline.output).toBeInstanceOf(Array)
+    // Bob should see posts from Alice and Carol (at least 4 posts)
+    expect(timeline.output.length).toBeGreaterThanOrEqual(4)
+  })
+
+  test("should return posts with correct structure", async () => {
+    const timeline = await Trotsky.init(agent)
+      .timeline()
+      .runHere()
+
+    timeline.output.forEach(post => {
+      expect(post).toHaveProperty("uri")
+      expect(post).toHaveProperty("cid")
+      expect(post).toHaveProperty("author")
+      expect(post).toHaveProperty("record")
+      expect(post.author).toHaveProperty("handle")
+      expect(post.author).toHaveProperty("did")
+    })
+  })
+
+  test("should verify timeline contains posts from followed users", async () => {
+    const timeline = await Trotsky.init(agent)
+      .timeline()
+      .runHere()
+
+    const authorDids = timeline.output.map(post => post.author.did)
+
+    // Timeline should contain posts from Alice or Carol
+    const hasAliceOrCarol = authorDids.some(did =>
+      did === alice.did || did === carol.did
+    )
+    expect(hasAliceOrCarol).toBe(true)
+  })
+
+  test("should iterate through each timeline post", async () => {
+    const postUris: string[] = []
+
+    await Trotsky.init(agent)
+      .timeline()
+      .take(3)
+      .each()
+      .tap((step) => {
+        if (step?.context?.uri) {
+          postUris.push(step.context.uri)
+        }
+      })
+      .run()
+
+    expect(postUris.length).toBeGreaterThan(0)
+    expect(postUris.length).toBeLessThanOrEqual(3)
+  })
+
+  test("should filter timeline posts with when()", async () => {
+    const filteredPosts: string[] = []
+
+    await Trotsky.init(agent)
+      .timeline()
+      .take(10)
+      .each()
+      .when((step) => step?.context?.author?.did === alice.did)
+      .tap((step) => {
+        if (step?.context?.uri) {
+          filteredPosts.push(step.context.uri)
+        }
+      })
+      .run()
+
+    // All filtered posts should be from Alice
+    const timeline = await Trotsky.init(agent).timeline().runHere()
+    const alicePosts = timeline.output.filter(p => p.author.did === alice.did)
+
+    expect(filteredPosts.length).toBeLessThanOrEqual(alicePosts.length)
+  })
+
+  test("should handle pagination with take()", async () => {
+    const timeline = await Trotsky.init(agent)
+      .timeline()
+      .take(2)
+      .runHere()
+
+    expect(timeline.output.length).toBeLessThanOrEqual(2)
+  })
+
+  test("should work with custom query parameters", async () => {
+    const timeline = await Trotsky.init(agent)
+      .timeline({ "limit": 5 })
+      .runHere()
+
+    expect(timeline.output).toBeInstanceOf(Array)
+    expect(timeline.output.length).toBeLessThanOrEqual(5)
+  })
+
+  test("should return indexed timestamps", async () => {
+    const timeline = await Trotsky.init(agent)
+      .timeline()
+      .take(5)
+      .runHere()
+
+    timeline.output.forEach(post => {
+      expect(post).toHaveProperty("indexedAt")
+      expect(post.indexedAt).toBeTruthy()
+    })
+  })
+
+  test("should work with tap() to process results", async () => {
+    let processedCount = 0
+
+    await Trotsky.init(agent)
+      .timeline()
+      .take(3)
+      .each()
+      .tap(() => {
+        processedCount++
+      })
+      .run()
+
+    expect(processedCount).toBeGreaterThan(0)
+    expect(processedCount).toBeLessThanOrEqual(3)
+  })
+
+  test("should handle empty timeline for new user", async () => {
+    // Create a new user who doesn't follow anyone
+    await sc.createAccount("newuser", {
+      "handle": "newuser.test",
+      "email": "newuser@test.com",
+      "password": "password"
+    })
+
+    const newAgent = network.pds.getClient()
+    await newAgent.login({ "identifier": "newuser.test", "password": "password" })
+
+    await network.processAll()
+
+    const timeline = await Trotsky.init(newAgent)
+      .timeline()
+      .runHere()
+
+    expect(timeline.output).toBeInstanceOf(Array)
+    // New user should have empty or minimal timeline
+    expect(timeline.output.length).toBeLessThanOrEqual(10)
+  })
+
+  test("should support algorithm parameter", async () => {
+    const timeline = await Trotsky.init(agent)
+      .timeline({ "algorithm": "reverse-chronological" })
+      .take(5)
+      .runHere()
+
+    expect(timeline.output).toBeInstanceOf(Array)
+  })
+})