feat: implement matrix adapter fetch.

Author: kcexn

internal/email/adapter_fetch.go

@@ -57,7 +57,7 @@ func (a *Adapter) fetchMessages(ctx context.Context, startUID uint32) ([]relay.M
 func (a *Adapter) getCursor(ctx context.Context) (uint32, uint32, error) {
 	var uid, uidValidity uint32
 	err := store.Client().WithTx(ctx, func(ctx context.Context, tx store.Tx) error {
-		u, v, err := tx.Cursor(ctx, a.cfg.Username, a.cfg.Mailbox)
+		u, v, err := tx.Email().Cursor(ctx, a.cfg.Username, a.cfg.Mailbox)
 		if err != nil {
 			return err
 		}
@@ -71,7 +71,7 @@ func (a *Adapter) getCursor(ctx context.Context) (uint32, uint32, error) {
 // setCursor durably persists the given UID and UIDValidity to the store.
 func (a *Adapter) setCursor(ctx context.Context, uid, uidValidity uint32) error {
 	return store.Client().WithTx(ctx, func(ctx context.Context, tx store.Tx) error {
-		return tx.SetCursor(ctx, a.cfg.Username, a.cfg.Mailbox, uid, uidValidity)
+		return tx.Email().SetCursor(ctx, a.cfg.Username, a.cfg.Mailbox, uid, uidValidity)
 	})
 }
 

internal/email/adapter_test.go

@@ -61,7 +61,7 @@ func TestGetCursor(t *testing.T) {
 	// Seed the cursor through the store directly, so the test exercises only the
 	// read path.
 	err := store.Client().WithTx(ctx, func(ctx context.Context, tx store.Tx) error {
-		return tx.SetCursor(ctx, a.cfg.Username, a.cfg.Mailbox, wantUID, wantValidity)
+		return tx.Email().SetCursor(ctx, a.cfg.Username, a.cfg.Mailbox, wantUID, wantValidity)
 	})
 	if err != nil {
 		t.Fatalf("seed cursor: %v", err)
@@ -96,7 +96,7 @@ func TestSetCursor(t *testing.T) {
 	var gotUID, gotValidity uint32
 	err := store.Client().WithTx(ctx, func(ctx context.Context, tx store.Tx) error {
 		var err error
-		gotUID, gotValidity, err = tx.Cursor(ctx, a.cfg.Username, a.cfg.Mailbox)
+		gotUID, gotValidity, err = tx.Email().Cursor(ctx, a.cfg.Username, a.cfg.Mailbox)
 		return err
 	})
 	if err != nil {

internal/matrix/adapter.go

@@ -7,15 +7,15 @@ import (
 	"bolte-bridge/internal/relay"
 )
 
-// Adapter is the Matrix medium edge of the bridge. It owns the Client it talks
-// to outright: the transport is an implementation detail of this package, so
-// callers configure the adapter and never name the Client themselves.
-//
-// Its Fetch, Send, and Commit methods are not implemented yet.
+// Adapter is the Matrix medium edge of the bridge.
 type Adapter struct {
 	client Client
+	cfg    Config
 }
 
+// Compile-time assertion that Adapter satisfies core.Adapter.
+var _ core.Adapter = (*Adapter)(nil)
+
 // NewAdapter builds an Adapter and the Client it owns from ctx and cfg,
 // reporting any configuration or initialization error from Client construction.
 func NewAdapter(ctx context.Context, cfg Config) (*Adapter, error) {
@@ -34,27 +34,24 @@ func (a *Adapter) Medium() relay.Medium {
 	return relay.MediumMatrix
 }
 
-// Fetch will fetch Matrix events past the committed sync token and translate
+// Fetch will fetch Matrix events past the committed EventID and translate
 // them into relay messages, advancing only the in-memory cursor.
-func (a *Adapter) Fetch(_ context.Context) ([]relay.Message, error) {
-	return nil, nil
+func (a *Adapter) Fetch(ctx context.Context) ([]relay.Message, error) {
+	return a.fetch(ctx)
 }
 
 // Send will translate a routed message into a Matrix event, send it to the
-// configured room, and return the event ID assigned by the homeserver.
+// configured room, and return the EventID assigned by the homeserver.
 func (a *Adapter) Send(_ context.Context, _ relay.RoutedMessage) (string, error) {
 	return "", nil
 }
 
-// Commit will durably advance the Matrix sync token to cursor. An empty cursor
+// Commit will durably advance the Matrix EventID cursor. An empty cursor
 // commits everything returned by the preceding Fetch.
 func (a *Adapter) Commit(_ context.Context, _ string) error {
 	return nil
 }
 
-// Compile-time assertion that Adapter satisfies core.Adapter.
-var _ core.Adapter = (*Adapter)(nil)
-
 // Close closes the underlying Matrix client.
 func (a *Adapter) Close(ctx context.Context) error {
 	return a.client.Close(ctx)

internal/matrix/adapter_fetch.go

@@ -0,0 +1,79 @@
+package matrix
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+
+	"bolte-bridge/internal/relay"
+	"bolte-bridge/internal/store"
+)
+
+func (a *Adapter) fetch(ctx context.Context) ([]relay.Message, error) {
+	eventID, err := a.getCursor(ctx)
+	switch {
+	case errors.Is(err, sql.ErrNoRows):
+		lastID, err := a.client.LastEvent(ctx)
+		if err != nil {
+			return nil, err
+		}
+		return nil, a.setCursor(ctx, lastID)
+	case err != nil:
+		return nil, err
+	default:
+		return a.fetchMessages(ctx, eventID)
+	}
+}
+
+// getCursor retrieves the current EventID from the store.
+func (a *Adapter) getCursor(ctx context.Context) (string, error) {
+	var eventID string
+	err := store.Client().WithTx(ctx, func(ctx context.Context, tx store.Tx) error {
+		e, err := tx.Matrix().Cursor(ctx, a.cfg.ServerName, a.cfg.RoomID)
+		if err != nil {
+			return err
+		}
+		eventID = e
+		return nil
+	})
+	return eventID, err
+}
+
+// setCursor retrieves the current EventID from the store.
+func (a *Adapter) setCursor(ctx context.Context, eventID string) error {
+	return store.Client().WithTx(ctx, func(ctx context.Context, tx store.Tx) error {
+		return tx.Matrix().SetCursor(ctx, a.cfg.ServerName, a.cfg.RoomID, eventID)
+	})
+}
+
+func (a *Adapter) fetchMessages(ctx context.Context, eventID string) ([]relay.Message, error) {
+	rawEvents, err := a.client.Fetch(ctx, eventID)
+	if err != nil {
+		return nil, err
+	}
+
+	return rawEventsToRelayMessages(rawEvents), nil
+}
+
+func rawEventsToRelayMessages(
+	rawEvents []RawEvent,
+) []relay.Message {
+	messages := make([]relay.Message, len(rawEvents))
+
+	for i, raw := range rawEvents {
+		messages[i] = rawEventToRelayMessage(raw)
+	}
+	return messages
+}
+
+func rawEventToRelayMessage(raw RawEvent) relay.Message {
+	return relay.Message{
+		Sender: relay.Identity{Address: relay.Address{
+			Mode: relay.MediumMatrix,
+			ID:   raw.Sender,
+		}},
+		MessageID: raw.EventID,
+		ThreadID:  raw.ReplyTo,
+		Body:      raw.Body,
+	}
+}