xrpc: add place.stream.live.getSegments

js/app/features/bluesky/blueskySlice.tsx

···
 
         console.log("getting profiles");
         return await bskyAgent.getProfiles({
-          actors: actors,
+          actors: [actors.join(",")],
         });
       },
       {

js/docs/src/content/docs/lex-reference/live/place-stream-live-getsegments.md

···
+---
+title: place.stream.live.getSegments
+description: Reference for the place.stream.live.getSegments lexicon
+---
+
+**Lexicon Version:** 1
+
+## Definitions
+
+<a name="main"></a>
+
+### `main`
+
+**Type:** `query`
+
+Get a list of livestream segments for a user
+
+**Parameters:**
+
+| Name      | Type      | Req'd | Description                               | Constraints                           |
+| --------- | --------- | ----- | ----------------------------------------- | ------------------------------------- |
+| `userDID` | `string`  | ✅    | The DID of the potentially-following user | Format: `did`                         |
+| `limit`   | `integer` | ❌    |                                           | Min: 1<br/>Max: 100<br/>Default: `50` |
+| `before`  | `string`  | ❌    |                                           | Format: `datetime`                    |
+
+**Output:**
+
+- **Encoding:** `application/json`
+- **Schema:**
+
+**Schema Type:** `object`
+
+| Name       | Type                                                                                           | Req'd | Description | Constraints |
+| ---------- | ---------------------------------------------------------------------------------------------- | ----- | ----------- | ----------- |
+| `segments` | Array of [`place.stream.segment#segmentView`](/lex-reference/place-stream-segment#segmentview) | ❌    |             |             |
+
+---
+
+## Lexicon Source
+
+```json
+{
+  "lexicon": 1,
+  "id": "place.stream.live.getSegments",
+  "defs": {
+    "main": {
+      "type": "query",
+      "description": "Get a list of livestream segments for a user",
+      "parameters": {
+        "type": "params",
+        "required": ["userDID"],
+        "properties": {
+          "userDID": {
+            "type": "string",
+            "format": "did",
+            "description": "The DID of the potentially-following user"
+          },
+          "limit": {
+            "type": "integer",
+            "minimum": 1,
+            "maximum": 100,
+            "default": 50
+          },
+          "before": {
+            "type": "string",
+            "format": "datetime"
+          }
+        }
+      },
+      "output": {
+        "encoding": "application/json",
+        "schema": {
+          "type": "object",
+          "properties": {
+            "segments": {
+              "type": "array",
+              "items": {
+                "type": "ref",
+                "ref": "place.stream.segment#segmentView"
+              }
+            }
+          }
+        }
+      }
+    }
+  }
+}
+```

js/docs/src/content/docs/lex-reference/place-stream-segment.md

···
 
 ---
 
+<a name="segmentview"></a>
+
+### `segmentView`
+
+**Type:** `object`
+
+**Properties:**
+
+| Name     | Type      | Req'd | Description | Constraints   |
+| -------- | --------- | ----- | ----------- | ------------- |
+| `cid`    | `string`  | ✅    |             | Format: `cid` |
+| `record` | `unknown` | ✅    |             |               |
+
+---
+
 ## Lexicon Source
 
 ```json
···
         },
         "den": {
           "type": "integer"
+        }
+      }
+    },
+    "segmentView": {
+      "type": "object",
+      "required": ["cid", "record"],
+      "properties": {
+        "cid": {
+          "type": "string",
+          "format": "cid"
+        },
+        "record": {
+          "type": "unknown"
         }
       }
     }

lexicons/place/stream/live/getSegments.json

···
+{
+  "lexicon": 1,
+  "id": "place.stream.live.getSegments",
+  "defs": {
+    "main": {
+      "type": "query",
+      "description": "Get a list of livestream segments for a user",
+      "parameters": {
+        "type": "params",
+        "required": ["userDID"],
+        "properties": {
+          "userDID": {
+            "type": "string",
+            "format": "did",
+            "description": "The DID of the potentially-following user"
+          },
+          "limit": {
+            "type": "integer",
+            "minimum": 1,
+            "maximum": 100,
+            "default": 50
+          },
+          "before": { "type": "string", "format": "datetime" }
+        }
+      },
+      "output": {
+        "encoding": "application/json",
+        "schema": {
+          "type": "object",
+          "properties": {
+            "segments": {
+              "type": "array",
+              "items": {
+                "type": "ref",
+                "ref": "place.stream.segment#segmentView"
+              }
+            }
+          }
+        }
+      }
+    }
+  }
+}

lexicons/place/stream/segment.json

···
         "num": { "type": "integer" },
         "den": { "type": "integer" }
       }
+    },
+    "segmentView": {
+      "type": "object",
+      "required": ["cid", "record"],
+      "properties": {
+        "cid": { "type": "string", "format": "cid" },
+        "record": { "type": "unknown" }
+      }
     }
   }
 }

pkg/model/model.go

···
 	MostRecentSegments() ([]Segment, error)
 	MostRecentSegmentsWithStreamInfo() ([]SegmentWithStreamInfo, error)
 	LatestSegmentForUser(user string) (*Segment, error)
+	LatestSegmentsForUser(user string, limit int, before *time.Time) ([]Segment, error)
 	CreateThumbnail(thumb *Thumbnail) error
 	LatestThumbnailForUser(user string) (*Thumbnail, error)
 	GetSegment(id string) (*Segment, error)

pkg/model/segment.go

···
 	return &seg, nil
 }
 
+func (m *DBModel) LatestSegmentsForUser(user string, limit int, before *time.Time) ([]Segment, error) {
+	var segs []Segment
+	if before == nil {
+		later := time.Now().Add(1000 * time.Hour)
+		before = &later
+	}
+	err := m.DB.Model(Segment{}).Where("repo_did = ? AND start_time < ?", user, before.UTC()).Order("start_time DESC").Limit(limit).Find(&segs).Error
+	if err != nil {
+		return nil, err
+	}
+	return segs, nil
+}
+
 func (m *DBModel) GetLiveUsers() ([]Segment, error) {
 	var liveUsers []Segment
 	thirtySecondsAgo := aqtime.FromTime(time.Now().Add(-30 * time.Second)).Time()
···
 	// finally, make the resulting SegmentWithStreamInfos
 	resultWithStreamInfo := make([]SegmentWithStreamInfo, len(recentSegments))
 	for i, seg := range recentSegments {
+		view, ok := livestreamMap[seg.RepoDID]
+		if !ok {
+			log.Error(context.Background(), "No livestream view found for repo_did", "repo_did", seg.RepoDID)
+			continue
+		}
 		resultWithStreamInfo[i] = SegmentWithStreamInfo{
 			Segment:        seg,
-			LivestreamView: livestreamMap[seg.RepoDID].Record,
+			LivestreamView: view.Record,
 		}
 	}
 

pkg/spxrpc/place_stream_live.go

···
+package spxrpc
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/bluesky-social/indigo/lex/util"
+	"github.com/bluesky-social/indigo/repo"
+	"github.com/ipfs/go-cid"
+	"github.com/labstack/echo/v4"
+	"github.com/multiformats/go-multihash"
+	placestreamtypes "stream.place/streamplace/pkg/streamplace"
+)
+
+func (s *Server) handlePlaceStreamLiveGetSegments(ctx context.Context, before string, limit int, userDID string) (*placestreamtypes.LiveGetSegments_Output, error) {
+	var beforeTime *time.Time
+	if before != "" {
+		parsedTime, err := time.Parse(time.RFC3339, before)
+		if err != nil {
+			return nil, echo.NewHTTPError(http.StatusBadRequest, "Invalid 'before' parameter: must be RFC3339 format")
+		}
+		beforeTime = &parsedTime
+	}
+
+	segments, err := s.model.LatestSegmentsForUser(userDID, limit, beforeTime)
+	if err != nil {
+		return nil, echo.NewHTTPError(http.StatusInternalServerError, "Failed to fetch segments")
+	}
+
+	// Convert segments to the expected output format
+	output := &placestreamtypes.LiveGetSegments_Output{
+		Segments: make([]*placestreamtypes.Segment_SegmentView, len(segments)),
+	}
+
+	for i, segment := range segments {
+		record, err := segment.ToStreamplaceSegment()
+		if err != nil {
+			return nil, echo.NewHTTPError(http.StatusInternalServerError, fmt.Sprintf("Failed to convert segment to streamplace segment: %s", err))
+		}
+		c, err := getCID(record)
+		if err != nil {
+			return nil, echo.NewHTTPError(http.StatusInternalServerError, fmt.Sprintf("Failed to get CID: %s", err))
+		}
+		ltd := &util.LexiconTypeDecoder{Val: record}
+
+		output.Segments[i] = &placestreamtypes.Segment_SegmentView{
+			Record: ltd,
+			Cid:    c.String(),
+		}
+	}
+
+	return output, nil
+}
+
+func getCID(record repo.CborMarshaler) (*cid.Cid, error) {
+	builder := cid.NewPrefixV1(cid.DagCBOR, multihash.SHA2_256)
+	buf := bytes.NewBuffer(nil)
+	err := record.MarshalCBOR(buf)
+	if err != nil {
+		return nil, err
+	}
+	c, err := builder.Sum(buf.Bytes())
+	if err != nil {
+		return nil, err
+	}
+	return &c, nil
+}

pkg/spxrpc/stubs.go

···
 
 func (s *Server) RegisterHandlersPlaceStream(e *echo.Echo) error {
 	e.GET("/xrpc/place.stream.graph.getFollowingUser", s.HandlePlaceStreamGraphGetFollowingUser)
+	e.GET("/xrpc/place.stream.live.getSegments", s.HandlePlaceStreamLiveGetSegments)
 	return nil
 }
 
···
 	var handleErr error
 	// func (s *Server) handlePlaceStreamGraphGetFollowingUser(ctx context.Context,subjectDID string,userDID string) (*placestreamtypes.GraphGetFollowingUser_Output, error)
 	out, handleErr = s.handlePlaceStreamGraphGetFollowingUser(ctx, subjectDID, userDID)
+	if handleErr != nil {
+		return handleErr
+	}
+	return c.JSON(200, out)
+}
+
+func (s *Server) HandlePlaceStreamLiveGetSegments(c echo.Context) error {
+	ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamLiveGetSegments")
+	defer span.End()
+	before := c.QueryParam("before")
+
+	var limit int
+	if p := c.QueryParam("limit"); p != "" {
+		var err error
+		limit, err = strconv.Atoi(p)
+		if err != nil {
+			return err
+		}
+	} else {
+		limit = 50
+	}
+	userDID := c.QueryParam("userDID")
+	var out *placestreamtypes.LiveGetSegments_Output
+	var handleErr error
+	// func (s *Server) handlePlaceStreamLiveGetSegments(ctx context.Context,before string,limit int,userDID string) (*placestreamtypes.LiveGetSegments_Output, error)
+	out, handleErr = s.handlePlaceStreamLiveGetSegments(ctx, before, limit, userDID)
 	if handleErr != nil {
 		return handleErr
 	}

pkg/streamplace/livegetSegments.go

···
+// Code generated by cmd/lexgen (see Makefile's lexgen); DO NOT EDIT.
+
+package streamplace
+
+// schema: place.stream.live.getSegments
+
+import (
+	"context"
+
+	"github.com/bluesky-social/indigo/xrpc"
+)
+
+// LiveGetSegments_Output is the output of a place.stream.live.getSegments call.
+type LiveGetSegments_Output struct {
+	Segments []*Segment_SegmentView `json:"segments,omitempty" cborgen:"segments,omitempty"`
+}
+
+// LiveGetSegments calls the XRPC method "place.stream.live.getSegments".
+//
+// userDID: The DID of the potentially-following user
+func LiveGetSegments(ctx context.Context, c *xrpc.Client, before string, limit int64, userDID string) (*LiveGetSegments_Output, error) {
+	var out LiveGetSegments_Output
+
+	params := map[string]interface{}{
+		"before":  before,
+		"limit":   limit,
+		"userDID": userDID,
+	}
+	if err := c.Do(ctx, xrpc.Query, "", "place.stream.live.getSegments", params, nil, &out); err != nil {
+		return nil, err
+	}
+
+	return &out, nil
+}

pkg/streamplace/streamsegment.go

···
 	Num int64 `json:"num" cborgen:"num"`
 }
 
+// Segment_SegmentView is a "segmentView" in the place.stream.segment schema.
+type Segment_SegmentView struct {
+	Cid    string                   `json:"cid" cborgen:"cid"`
+	Record *util.LexiconTypeDecoder `json:"record" cborgen:"record"`
+}
+
 // Segment_Video is a "video" in the place.stream.segment schema.
 type Segment_Video struct {
 	Codec     string             `json:"codec" cborgen:"codec"`