diff --git a/blob/integration.go b/blob/integration.go new file mode 100644 index 0000000..97bc97c --- /dev/null +++ b/blob/integration.go @@ -0,0 +1,124 @@ +package blob + +import ( + "bytes" + "context" + "errors" + + blobpb "github.com/code-payments/flipcash2-protobuf-api/generated/go/blob/v1" + commonpb "github.com/code-payments/flipcash2-protobuf-api/generated/go/common/v1" +) + +// ErrBlobNotShareable is returned by Integration.ShareIntoChat when a referenced +// blob cannot be attached to a chat — it does not exist, is not owned by the +// sharer, is not a READY original, or is not an image. When it is returned none of +// the blobs are granted. +var ErrBlobNotShareable = errors.New("blob not shareable") + +// Integration is the surface other domains (messaging today) use to attach blobs +// to a resource they own: it validates and grants read access on send +// (ShareIntoChat), and resolves the blobs' metadata on read (Resolve). +type Integration struct { + blobs Store + storage ObjectStorage + access AccessStore +} + +// NewIntegration returns an Integration backed by the given blob metadata store, +// object storage, and ACL store. +func NewIntegration(blobs Store, storage ObjectStorage, access AccessStore) *Integration { + return &Integration{blobs: blobs, storage: storage, access: access} +} + +// ShareIntoChat attaches blobs to a chat: it verifies that sharerID owns every +// blob in blobIDs and that each is a READY image original, then grants the chat +// read access to each. It is all-or-nothing — if any blob fails validation +// nothing is granted and ErrBlobNotShareable is returned — and idempotent, so a +// re-sent message re-grants harmlessly. An empty blobIDs is a no-op. +// +// Only the owner may introduce a blob into a chat: a BlobId is a bearer +// capability, so without the ownership check a member could attach a blob they +// merely learned the id of. Only a READY original is servable and grantable +// (renditions inherit their original's grants), so a pending, rejected, or +// rendition blob is rejected; and chat media is images only today, so a +// non-image blob is rejected too. +func (i *Integration) ShareIntoChat(ctx context.Context, sharerID *commonpb.UserId, chatID *commonpb.ChatId, blobIDs []*blobpb.BlobId) error { + if len(blobIDs) == 0 { + return nil + } + + records, err := i.blobs.GetByIDs(ctx, blobIDs) + if err != nil { + return err + } + byID := make(map[string]*Blob, len(records)) + for _, b := range records { + byID[string(b.ID.Value)] = b + } + + // Validate every requested blob before granting any, so a single bad id in the + // batch grants nothing. + for _, id := range blobIDs { + b, ok := byID[string(id.Value)] + if !ok { + return ErrBlobNotShareable + } + if b.Owner == nil || !bytes.Equal(b.Owner.Value, sharerID.Value) { + return ErrBlobNotShareable + } + if b.Rendition != RenditionOriginal || b.ParentID != nil { + return ErrBlobNotShareable + } + // Media in a chat is images only today; reject any other content kind (a + // future video/file blob) rather than attaching something the chat cannot + // render or serve. + if !SupportedImageMimeTypes[b.MimeType] { + return ErrBlobNotShareable + } + if b.State != StateReady { + return ErrBlobNotShareable + } + } + + chat := PrincipalForChat(chatID) + for _, id := range blobIDs { + if err := i.access.Grant(ctx, &Grant{BlobID: id, Principal: chat, Permission: PermissionRead}); err != nil { + return err + } + } + return nil +} + +// Resolve returns the server-authoritative metadata — mime type, size, image +// dimensions, and a freshly minted, short-lived download URL — for each READY +// blob among ids, keyed by string(BlobId.Value). Unknown or not-yet-READY ids are +// omitted. An empty input yields a nil map. +// +// It performs NO authorization. Callers must only pass ids they have already +// established the reader may see — e.g. blob ids drawn from a chat message the +// reader is a member of, which were granted to the chat when the message was sent. +func (i *Integration) Resolve(ctx context.Context, ids []*blobpb.BlobId) (map[string]*blobpb.BlobMetadata, error) { + if len(ids) == 0 { + return nil, nil + } + + records, err := i.blobs.GetByIDs(ctx, ids) + if err != nil { + return nil, err + } + + out := make(map[string]*blobpb.BlobMetadata, len(records)) + for _, record := range records { + // Only a READY blob is servable; a pending/rejected one has no metadata to + // surface, so it is left for the client to treat as unavailable. + if record.State != StateReady { + continue + } + metadata, err := buildMetadata(ctx, i.storage, record) + if err != nil { + return nil, err + } + out[string(record.ID.Value)] = metadata + } + return out, nil +} diff --git a/blob/integration_test.go b/blob/integration_test.go new file mode 100644 index 0000000..7782024 --- /dev/null +++ b/blob/integration_test.go @@ -0,0 +1,157 @@ +package blob_test + +import ( + "context" + "testing" + + "github.com/google/uuid" + "github.com/stretchr/testify/require" + + blobpb "github.com/code-payments/flipcash2-protobuf-api/generated/go/blob/v1" + commonpb "github.com/code-payments/flipcash2-protobuf-api/generated/go/common/v1" + + "github.com/code-payments/flipcash2-server/blob" + "github.com/code-payments/flipcash2-server/blob/memory" + "github.com/code-payments/flipcash2-server/model" +) + +func newBlobID(t *testing.T) *blobpb.BlobId { + id, err := uuid.NewRandom() + require.NoError(t, err) + v := id + return &blobpb.BlobId{Value: v[:]} +} + +func newChatID() *commonpb.ChatId { + a := model.MustGenerateUserID().Value + b := model.MustGenerateUserID().Value + return &commonpb.ChatId{Value: append(append([]byte{}, a...), b...)} +} + +func putReadyOriginal(t *testing.T, store blob.Store, owner *commonpb.UserId) *blobpb.BlobId { + ctx := context.Background() + id := newBlobID(t) + require.NoError(t, store.CreatePending(ctx, &blob.Blob{ + ID: id, + Rendition: blob.RenditionOriginal, + Owner: owner, + State: blob.StatePending, + StorageKey: "images/x/original.png", + MimeType: "image/png", + SizeBytes: 1, + })) + _, err := store.Advance(ctx, id, blob.StateReady, nil) + require.NoError(t, err) + return id +} + +func TestIntegration_ShareIntoChat(t *testing.T) { + ctx := context.Background() + store := memory.NewInMemory() + access := memory.NewInMemoryAccessStore() + integration := blob.NewIntegration(store, memory.NewInMemoryStorage(), access) + + owner := model.MustGenerateUserID() + chatID := newChatID() + chatPrincipal := blob.PrincipalForChat(chatID) + + t.Run("a ready original owned by the sharer is shared and granted", func(t *testing.T) { + id := putReadyOriginal(t, store, owner) + require.NoError(t, integration.ShareIntoChat(ctx, owner, chatID, []*blobpb.BlobId{id})) + + has, err := access.HasGrant(ctx, id, chatPrincipal, blob.PermissionRead) + require.NoError(t, err) + require.True(t, has) + + // Idempotent: sharing again leaves the grant in place, not an error. + require.NoError(t, integration.ShareIntoChat(ctx, owner, chatID, []*blobpb.BlobId{id})) + }) + + t.Run("empty input is a no-op", func(t *testing.T) { + require.NoError(t, integration.ShareIntoChat(ctx, owner, chatID, nil)) + }) + + t.Run("a blob owned by someone else is rejected and grants nothing", func(t *testing.T) { + mine := putReadyOriginal(t, store, owner) + theirs := putReadyOriginal(t, store, model.MustGenerateUserID()) + + err := integration.ShareIntoChat(ctx, owner, chatID, []*blobpb.BlobId{mine, theirs}) + require.ErrorIs(t, err, blob.ErrBlobNotShareable) + + // All-or-nothing: the valid blob in the batch was not granted either. + has, err := access.HasGrant(ctx, mine, chatPrincipal, blob.PermissionRead) + require.NoError(t, err) + require.False(t, has) + }) + + t.Run("a non-ready blob is rejected", func(t *testing.T) { + id := newBlobID(t) + require.NoError(t, store.CreatePending(ctx, &blob.Blob{ + ID: id, Rendition: blob.RenditionOriginal, Owner: owner, State: blob.StatePending, + StorageKey: "k", MimeType: "image/png", SizeBytes: 1, + })) + require.ErrorIs(t, integration.ShareIntoChat(ctx, owner, chatID, []*blobpb.BlobId{id}), blob.ErrBlobNotShareable) + }) + + t.Run("a rendition is rejected (only originals are shareable)", func(t *testing.T) { + original := putReadyOriginal(t, store, owner) + rendition := newBlobID(t) + require.NoError(t, store.CreatePending(ctx, &blob.Blob{ + ID: rendition, Rendition: blob.RenditionDisplay, ParentID: original, Owner: owner, State: blob.StatePending, + StorageKey: "k", MimeType: "image/png", SizeBytes: 1, + })) + _, err := store.Advance(ctx, rendition, blob.StateReady, nil) + require.NoError(t, err) + require.ErrorIs(t, integration.ShareIntoChat(ctx, owner, chatID, []*blobpb.BlobId{rendition}), blob.ErrBlobNotShareable) + }) + + t.Run("a non-image blob is rejected", func(t *testing.T) { + id := newBlobID(t) + require.NoError(t, store.CreatePending(ctx, &blob.Blob{ + ID: id, Rendition: blob.RenditionOriginal, Owner: owner, State: blob.StatePending, + StorageKey: "k", MimeType: "video/mp4", SizeBytes: 1, + })) + _, err := store.Advance(ctx, id, blob.StateReady, nil) + require.NoError(t, err) + require.ErrorIs(t, integration.ShareIntoChat(ctx, owner, chatID, []*blobpb.BlobId{id}), blob.ErrBlobNotShareable) + }) + + t.Run("an unknown blob is rejected", func(t *testing.T) { + require.ErrorIs(t, integration.ShareIntoChat(ctx, owner, chatID, []*blobpb.BlobId{newBlobID(t)}), blob.ErrBlobNotShareable) + }) +} + +func TestIntegration_Resolve(t *testing.T) { + ctx := context.Background() + store := memory.NewInMemory() + integration := blob.NewIntegration(store, memory.NewInMemoryStorage(), memory.NewInMemoryAccessStore()) + + owner := model.MustGenerateUserID() + + // A READY blob resolves to its metadata with a fresh download URL. + ready := putReadyOriginal(t, store, owner) + // A pending (non-READY) blob is omitted. + pending := newBlobID(t) + require.NoError(t, store.CreatePending(ctx, &blob.Blob{ + ID: pending, Rendition: blob.RenditionOriginal, Owner: owner, State: blob.StatePending, + StorageKey: "k", MimeType: "image/png", SizeBytes: 1, + })) + // An unknown id is omitted. + unknown := newBlobID(t) + + meta, err := integration.Resolve(ctx, []*blobpb.BlobId{ready, pending, unknown}) + require.NoError(t, err) + require.Len(t, meta, 1) + + m := meta[string(ready.Value)] + require.NotNil(t, m) + require.Equal(t, "image/png", m.MimeType) + require.EqualValues(t, 1, m.SizeBytes) + require.NotNil(t, m.DownloadUrl) + require.NotEmpty(t, m.DownloadUrl.Url) + + // Empty input yields a nil map. + empty, err := integration.Resolve(ctx, nil) + require.NoError(t, err) + require.Nil(t, empty) +} diff --git a/blob/model.go b/blob/model.go index e35536f..0612bc6 100644 --- a/blob/model.go +++ b/blob/model.go @@ -240,16 +240,16 @@ type Blob struct { Rejection *RejectionMetadata } -func newBlobID() (*blobpb.BlobId, error) { +func MustGenerateID() *blobpb.BlobId { id, err := uuid.NewRandom() if err != nil { - return nil, err + panic(err) } value := id - return &blobpb.BlobId{Value: value[:]}, nil + return &blobpb.BlobId{Value: value[:]} } -func blobIDString(id *blobpb.BlobId) string { +func IDString(id *blobpb.BlobId) string { if id == nil { return "" } diff --git a/blob/server.go b/blob/server.go index 30966ab..b868060 100644 --- a/blob/server.go +++ b/blob/server.go @@ -146,11 +146,9 @@ func (s *Server) InitiateExternalUpload(ctx context.Context, req *blobpb.Initiat }, nil } - id, err := newBlobID() - if err != nil { - log.Warn("Failed to generate blob id", zap.Error(err)) - return nil, status.Error(codes.Internal, "failed to initiate upload") - } + id := MustGenerateID() + log = log.With(zap.String("blob_id", IDString(id))) + // The mime type was validated as a supported image above, so this resolves; an // error here would mean the two lists drifted out of sync. key, err := StorageKey(id, req.MimeType) @@ -222,7 +220,7 @@ func (s *Server) CompleteExternalUpload(ctx context.Context, req *blobpb.Complet log := s.log.With( zap.String("owner_id", model.UserIDString(owner)), - zap.String("blob_id", blobIDString(req.BlobId)), + zap.String("blob_id", IDString(req.BlobId)), ) record, err := s.blobs.GetByID(ctx, req.BlobId) @@ -306,7 +304,7 @@ func (s *Server) GetBlobs(ctx context.Context, req *blobpb.GetBlobsRequest) (*bl allowed, err := s.canRead(ctx, caller, record, req.Context) if err != nil { s.log.Warn("Failed to authorize blob", - zap.String("blob_id", blobIDString(record.ID)), + zap.String("blob_id", IDString(record.ID)), zap.Error(err), ) return nil, status.Error(codes.Internal, "failed to get blobs") @@ -327,10 +325,10 @@ func (s *Server) GetBlobs(ctx context.Context, req *blobpb.GetBlobsRequest) (*bl case blobpb.BlobStatus_BLOB_STATUS_READY: // download_url and the rest of the metadata are only meaningful, and only // minted, for a servable (READY) blob. - metadata, err := s.buildMetadata(ctx, record) + metadata, err := buildMetadata(ctx, s.storage, record) if err != nil { s.log.Warn("Failed to mint blob metadata", - zap.String("blob_id", blobIDString(record.ID)), + zap.String("blob_id", IDString(record.ID)), zap.Error(err), ) return nil, status.Error(codes.Internal, "failed to get blobs") @@ -595,16 +593,18 @@ func (s *Server) currentStatus(ctx context.Context, id *blobpb.BlobId) (blobpb.B func (s *Server) cleanupUpload(ctx context.Context, record *Blob) { if err := s.storage.DeleteUpload(ctx, record.StorageKey); err != nil { s.log.Warn("Failed to delete upload object after finalization", - zap.String("blob_id", blobIDString(record.ID)), + zap.String("blob_id", IDString(record.ID)), zap.Error(err), ) } } // buildMetadata assembles the server-authoritative metadata for a READY blob, -// minting a fresh, short-lived download URL. -func (s *Server) buildMetadata(ctx context.Context, record *Blob) (*blobpb.BlobMetadata, error) { - downloadURL, err := s.storage.SignDownloadURL(ctx, record.StorageKey) +// minting a fresh, short-lived download URL. It is a free function over an +// ObjectStorage so both the Server (GetBlobs) and Media (Resolve) can mint +// metadata from their own storage without duplicating the logic. +func buildMetadata(ctx context.Context, storage ObjectStorage, record *Blob) (*blobpb.BlobMetadata, error) { + downloadURL, err := storage.SignDownloadURL(ctx, record.StorageKey) if err != nil { return nil, err } diff --git a/blob/storage.go b/blob/storage.go index 4df69e7..777cbdc 100644 --- a/blob/storage.go +++ b/blob/storage.go @@ -85,5 +85,5 @@ func StorageKey(id *blobpb.BlobId, mimeType string) (string, error) { // An image kind with no registered extension means the two image maps drifted. return "", fmt.Errorf("missing extension for image mime type %q", mimeType) } - return fmt.Sprintf("images/%s/original%s", blobIDString(id), ext), nil + return fmt.Sprintf("images/%s/original%s", IDString(id), ext), nil } diff --git a/blob/tests/access.go b/blob/tests/access.go index bacc980..ec64c01 100644 --- a/blob/tests/access.go +++ b/blob/tests/access.go @@ -38,7 +38,7 @@ func newChatID(t *testing.T) *commonpb.ChatId { func testAccessGrantHasRevoke(t *testing.T, store blob.AccessStore) { ctx := context.Background() - blobID := newBlobID(t) + blobID := blob.MustGenerateID() chat := blob.PrincipalForChat(newChatID(t)) // Absent before granting. @@ -69,8 +69,8 @@ func testAccessGrantHasRevoke(t *testing.T, store blob.AccessStore) { func testAccessNoCollision(t *testing.T, store blob.AccessStore) { ctx := context.Background() - blobA := newBlobID(t) - blobB := newBlobID(t) + blobA := blob.MustGenerateID() + blobB := blob.MustGenerateID() chat := blob.PrincipalForChat(newChatID(t)) user := blob.PrincipalForUser(model.MustGenerateUserID()) @@ -108,7 +108,7 @@ func testAccessNoCollision(t *testing.T, store blob.AccessStore) { func testAccessValidation(t *testing.T, store blob.AccessStore) { ctx := context.Background() - blobID := newBlobID(t) + blobID := blob.MustGenerateID() chat := blob.PrincipalForChat(newChatID(t)) // An unknown permission is rejected, not silently stored. diff --git a/blob/tests/store.go b/blob/tests/store.go index 1fd802a..d37d6eb 100644 --- a/blob/tests/store.go +++ b/blob/tests/store.go @@ -4,7 +4,6 @@ import ( "context" "testing" - "github.com/google/uuid" "github.com/stretchr/testify/require" blobpb "github.com/code-payments/flipcash2-protobuf-api/generated/go/blob/v1" @@ -27,15 +26,8 @@ func RunStoreTests(t *testing.T, store blob.Store, teardown func()) { } } -func newBlobID(t *testing.T) *blobpb.BlobId { - id, err := uuid.NewRandom() - require.NoError(t, err) - value := id - return &blobpb.BlobId{Value: value[:]} -} - func pendingOriginal(t *testing.T) *blob.Blob { - id := newBlobID(t) + id := blob.MustGenerateID() key, err := blob.StorageKey(id, "image/png") require.NoError(t, err) return &blob.Blob{ @@ -52,7 +44,7 @@ func pendingOriginal(t *testing.T) *blob.Blob { func testStoreCreateAndGet(t *testing.T, store blob.Store) { ctx := context.Background() - _, err := store.GetByID(ctx, newBlobID(t)) + _, err := store.GetByID(ctx, blob.MustGenerateID()) require.ErrorIs(t, err, blob.ErrNotFound) original := pendingOriginal(t) @@ -75,7 +67,7 @@ func testStoreCreateAndGet(t *testing.T, store blob.Store) { second := pendingOriginal(t) require.NoError(t, store.CreatePending(ctx, second)) - found, err := store.GetByIDs(ctx, []*blobpb.BlobId{original.ID, newBlobID(t), second.ID}) + found, err := store.GetByIDs(ctx, []*blobpb.BlobId{original.ID, blob.MustGenerateID(), second.ID}) require.NoError(t, err) require.Len(t, found, 2) } @@ -83,7 +75,7 @@ func testStoreCreateAndGet(t *testing.T, store blob.Store) { func testStoreAdvance(t *testing.T, store blob.Store) { ctx := context.Background() - advanced, err := store.Advance(ctx, newBlobID(t), blob.StateUploaded, nil) + advanced, err := store.Advance(ctx, blob.MustGenerateID(), blob.StateUploaded, nil) require.ErrorIs(t, err, blob.ErrNotFound) require.False(t, advanced) @@ -145,7 +137,7 @@ func testStoreReject(t *testing.T, store blob.Store) { ctx := context.Background() // Rejecting an unknown blob reports not-found. - advanced, err := store.Reject(ctx, newBlobID(t), &blob.RejectionMetadata{Reason: blob.RejectionReasonCorrupt}) + advanced, err := store.Reject(ctx, blob.MustGenerateID(), &blob.RejectionMetadata{Reason: blob.RejectionReasonCorrupt}) require.ErrorIs(t, err, blob.ErrNotFound) require.False(t, advanced) diff --git a/messaging/media.go b/messaging/media.go new file mode 100644 index 0000000..55c9de0 --- /dev/null +++ b/messaging/media.go @@ -0,0 +1,171 @@ +package messaging + +import ( + "context" + "errors" + + "go.uber.org/zap" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + blobpb "github.com/code-payments/flipcash2-protobuf-api/generated/go/blob/v1" + commonpb "github.com/code-payments/flipcash2-protobuf-api/generated/go/common/v1" + messagingpb "github.com/code-payments/flipcash2-protobuf-api/generated/go/messaging/v1" + + "github.com/code-payments/flipcash2-server/blob" +) + +// Media is the blob-side integration messaging uses for media: it shares the +// blobs a message references into the chat on send (ShareIntoChat) and resolves +// their metadata on read (Resolve). It is implemented by blob.Integration. +// +// ShareIntoChat returns blob.ErrBlobNotShareable when a referenced blob may not be +// attached (unknown, not owned by the sender, or not a READY original), in which +// case nothing is granted. Resolve performs no authorization — the caller has +// already gated on chat membership — and returns metadata keyed by +// string(BlobId.Value), omitting unknown or not-yet-READY ids. +type Media interface { + ShareIntoChat(ctx context.Context, sharerID *commonpb.UserId, chatID *commonpb.ChatId, blobIDs []*blobpb.BlobId) error + Resolve(ctx context.Context, ids []*blobpb.BlobId) (map[string]*blobpb.BlobMetadata, error) +} + +// validClientMedia reports whether a media content is one a client may author: +// at least one item, each carrying exactly one ORIGINAL rendition identified by a +// blob id and no server-resolved metadata (the server fills Blob on read, never +// the client). Ownership and readiness of the blobs are checked separately, when +// they are shared into the chat. +func validClientMedia(media *messagingpb.MediaContent) bool { + if media == nil || len(media.Items) == 0 { + return false + } + for _, item := range media.Items { + if len(item.Renditions) != 1 { + return false + } + r := item.Renditions[0] + if r.Role != messagingpb.MediaItemRendition_ORIGINAL || r.BlobId == nil || r.Blob != nil { + return false + } + } + return true +} + +// shareMessageMedia grants the chat read access to the blobs referenced by +// content, after verifying the sender owns each and it is a READY original. It +// reports denied=true when the media is not shareable (the caller returns its +// DENIED result) and a ready-to-return Internal error on an unexpected failure. +// Non-media content is a no-op. It runs before the message is persisted and +// broadcast, so the grants are durable before any recipient can resolve them. +func (s *Server) shareMessageMedia(ctx context.Context, log *zap.Logger, senderID *commonpb.UserId, chatID *commonpb.ChatId, content []*messagingpb.Content) (denied bool, err error) { + blobIDs := mediaBlobIDs(content) + if len(blobIDs) == 0 { + return false, nil + } + switch err := s.media.ShareIntoChat(ctx, senderID, chatID, blobIDs); { + case errors.Is(err, blob.ErrBlobNotShareable): + return true, nil + case err != nil: + log.With(zap.Error(err)).Warn("Failure sharing media into chat") + return false, status.Error(codes.Internal, "") + } + return false, nil +} + +// mediaBlobIDs returns the blob ids a message references — whether the media is +// the message body or the body of a reply — or nil when there is no media. It +// assumes the content already passed clientAllowedContent. +func mediaBlobIDs(content []*messagingpb.Content) []*blobpb.BlobId { + if len(content) != 1 { + return nil + } + body := content[0] + if reply, ok := body.Type.(*messagingpb.Content_Reply); ok { + if len(reply.Reply.Content) != 1 { + return nil + } + body = reply.Reply.Content[0] + } + media, ok := body.Type.(*messagingpb.Content_Media) + if !ok { + return nil + } + var ids []*blobpb.BlobId + for _, item := range media.Media.Items { + for _, r := range item.Renditions { + if r.BlobId != nil { + ids = append(ids, r.BlobId) + } + } + } + return ids +} + +// hydrateMedia fills each media rendition's Blob with freshly resolved metadata +// (mime type, size, a short-lived download URL, dimensions). The caller has +// already gated on chat membership, and every blob in a stored message was granted +// to the chat when it was sent, so this resolves without re-checking the ACL. A +// blob that no longer resolves (e.g. a later takedown left it non-READY) is left +// with a nil Blob for the client to treat as unavailable. +func hydrateMedia(ctx context.Context, resolver Media, protos []*messagingpb.Message) error { + renditions := mediaRenditions(protos) + if len(renditions) == 0 { + return nil + } + + ids := make([]*blobpb.BlobId, 0, len(renditions)) + seen := make(map[string]struct{}, len(renditions)) + for _, r := range renditions { + key := string(r.BlobId.Value) + if _, ok := seen[key]; ok { + continue + } + seen[key] = struct{}{} + ids = append(ids, r.BlobId) + } + + metadata, err := resolver.Resolve(ctx, ids) + if err != nil { + return err + } + for _, r := range renditions { + if m, ok := metadata[string(r.BlobId.Value)]; ok { + r.Blob = m + } + } + return nil +} + +// mediaRenditions returns every media rendition carrying a blob id across the +// messages — whether the media is a message body or a reply body — so its Blob can +// be hydrated in place. +func mediaRenditions(protos []*messagingpb.Message) []*messagingpb.MediaItemRendition { + var out []*messagingpb.MediaItemRendition + for _, msg := range protos { + for _, c := range msg.Content { + media := mediaOf(c) + if media == nil { + continue + } + for _, item := range media.Items { + for _, r := range item.Renditions { + if r.BlobId != nil { + out = append(out, r) + } + } + } + } + } + return out +} + +// mediaOf returns the MediaContent a content element carries — directly or as a +// reply body — or nil if there is none. +func mediaOf(c *messagingpb.Content) *messagingpb.MediaContent { + switch t := c.Type.(type) { + case *messagingpb.Content_Media: + return t.Media + case *messagingpb.Content_Reply: + return mediaOf(t.Reply.Content[0]) + } + return nil +} diff --git a/messaging/sender.go b/messaging/sender.go index b3f5a4f..f08c797 100644 --- a/messaging/sender.go +++ b/messaging/sender.go @@ -43,6 +43,11 @@ type Sender struct { messages Store profiles profile.Store + // media resolves blob metadata so a broadcast new-message event carries the + // same resolved media a read would. Hydration is best-effort and a no-op for + // media-free sends (e.g. server-authored cash messages). + media Media + ocpData ocp_data.Provider pusher push.Pusher @@ -56,6 +61,7 @@ func NewSender( chats chat.Store, messages Store, profiles profile.Store, + media Media, ocpData ocp_data.Provider, pusher push.Pusher, eventBus *event.Bus[*commonpb.UserId, *eventpb.Event], @@ -66,6 +72,7 @@ func NewSender( chats: chats, messages: messages, profiles: profiles, + media: media, ocpData: ocpData, pusher: pusher, eventBus: eventBus, @@ -94,7 +101,7 @@ func (s *Sender) Send( content []*messagingpb.Content, clientMessageID *messagingpb.ClientMessageId, countsTowardUnread bool, -) (*Message, error) { +) (*messagingpb.Message, error) { log := s.log if senderID != nil { log = log.With(zap.String("user_id", model.UserIDString(senderID))) @@ -122,12 +129,22 @@ func (s *Sender) Send( return nil, status.Error(codes.Internal, "") } + // Build the message proto once and resolve its media metadata, so the proto + // returned to the caller (the SendMessage response) and the one broadcast to + // members carry the same hydrated message. Best-effort and a no-op for + // media-free sends; the message is already persisted, so a resolution failure + // just leaves Blob unset for clients to re-fetch. + msgProto := msg.ToProto() + if err := hydrateMedia(ctx, s.media, []*messagingpb.Message{msgProto}); err != nil { + log.With(zap.Error(err)).Warn("Failure resolving media metadata") + } + // A retried send (same client message ID) already ran every side effect when // the message was first persisted — most importantly the push to members. // Re-running them would duplicate notifications, so return the original // message and stop here. if !created { - return msg, nil + return msgProto, nil } // The message is now durable, and a retry skips the side effects below — so @@ -169,7 +186,7 @@ func (s *Sender) Send( // log (it is deprecated but still populated during the transition). The // sender's read pointer and the new last activity are only included when they // actually advanced — a no-op must not broadcast a stale pointer or timestamp. - msgProto := msg.ToProto() + // msgProto was already built and media-hydrated above. update := &eventpb.ChatUpdate{ NewMessages: &messagingpb.MessageBatch{Messages: []*messagingpb.Message{msgProto}}, Events: &messagingpb.EventBatch{Events: []*messagingpb.Event{NewMessageSentEvent(msgProto)}}, @@ -190,5 +207,5 @@ func (s *Sender) Send( // which case publishChatUpdate loads them itself). publishChatUpdate(ctx, log, s.badges, s.chats, s.profiles, s.ocpData, s.pusher, s.eventBus, chatID, update, nil, members) - return msg, nil + return msgProto, nil } diff --git a/messaging/server.go b/messaging/server.go index 0348605..f163284 100644 --- a/messaging/server.go +++ b/messaging/server.go @@ -41,6 +41,7 @@ type Server struct { chats chat.Store messages Store + media Media sender *Sender @@ -52,6 +53,7 @@ func NewServer( authz auth.Authorizer, chats chat.Store, messages Store, + media Media, sender *Sender, ) *Server { return &Server{ @@ -59,6 +61,7 @@ func NewServer( authz: authz, chats: chats, messages: messages, + media: media, sender: sender, } } @@ -86,9 +89,14 @@ func (s *Server) GetMessage(ctx context.Context, req *messagingpb.GetMessageRequ return nil, status.Error(codes.Internal, "") } + proto := msg.ToProto() + if err := hydrateMedia(ctx, s.media, []*messagingpb.Message{proto}); err != nil { + log.With(zap.Error(err)).Warn("Failure resolving media metadata") + } + return &messagingpb.GetMessageResponse{ Result: messagingpb.GetMessageResponse_OK, - Message: msg.ToProto(), + Message: proto, }, nil } @@ -129,6 +137,9 @@ func (s *Server) GetMessages(ctx context.Context, req *messagingpb.GetMessagesRe for i, m := range msgs { protos[i] = m.ToProto() } + if err := hydrateMedia(ctx, s.media, protos); err != nil { + log.With(zap.Error(err)).Warn("Failure resolving media metadata") + } return &messagingpb.GetMessagesResponse{ Result: messagingpb.GetMessagesResponse_OK, Messages: &messagingpb.MessageBatch{Messages: protos}, @@ -217,6 +228,9 @@ func (s *Server) GetDelta(req *messagingpb.GetDeltaRequest, stream messagingpb.M for i, m := range msgs { batch[i] = m.ToProto() } + if err := hydrateMedia(ctx, s.media, batch); err != nil { + log.With(zap.Error(err)).Warn("Failure resolving media metadata") + } resp.Messages = &messagingpb.MessageBatch{Messages: batch} } if err := stream.Send(resp); err != nil { @@ -263,14 +277,24 @@ func (s *Server) SendMessage(ctx context.Context, req *messagingpb.SendMessageRe } } - msg, err := s.sender.Send(ctx, req.ChatId, userID, req.Content, req.ClientMessageId, true) + // Share any media into the chat — validate the sender owns each blob and it is + // a READY original, then grant the chat read access — before the message is + // persisted and broadcast, so the grants are durable before any recipient can + // resolve the blobs. + if denied, err := s.shareMessageMedia(ctx, log, userID, req.ChatId, req.Content); err != nil { + return nil, err + } else if denied { + return &messagingpb.SendMessageResponse{Result: messagingpb.SendMessageResponse_DENIED}, nil + } + + msgProto, err := s.sender.Send(ctx, req.ChatId, userID, req.Content, req.ClientMessageId, true) if err != nil { return nil, err } return &messagingpb.SendMessageResponse{ Result: messagingpb.SendMessageResponse_OK, - Message: msg.ToProto(), + Message: msgProto, }, nil } @@ -333,6 +357,14 @@ func (s *Server) EditMessage(ctx context.Context, req *messagingpb.EditMessageRe } } + // New media on an edit is shared into the chat just like a send, before the + // edit is persisted and broadcast. + if denied, err := s.shareMessageMedia(ctx, log, userID, req.ChatId, req.Content); err != nil { + return nil, err + } else if denied { + return &messagingpb.EditMessageResponse{Result: messagingpb.EditMessageResponse_DENIED}, nil + } + now := time.Now().UTC() updated, err := s.messages.EditMessage(ctx, req.ChatId, req.MessageId, req.Content, now, req.ExpectedEventSequence) switch { @@ -358,6 +390,12 @@ func (s *Server) EditMessage(ctx context.Context, req *messagingpb.EditMessageRe // apply the edit live via the message_edited event, or pick it up on their next // history load. updatedProto := updated.ToProto() + // Resolve media metadata onto the edited message before it is broadcast and + // returned. Best-effort: the edit is already committed, so a resolution failure + // just leaves Blob unset for the client to re-fetch rather than failing the edit. + if err := hydrateMedia(ctx, s.media, []*messagingpb.Message{updatedProto}); err != nil { + log.With(zap.Error(err)).Warn("Failure resolving media metadata for edit") + } publishChatUpdate(ctx, log, s.sender.badges, s.sender.chats, s.sender.profiles, s.sender.ocpData, s.sender.pusher, s.sender.eventBus, req.ChatId, &eventpb.ChatUpdate{ Events: &messagingpb.EventBatch{Events: []*messagingpb.Event{NewMessageEditedEvent(updatedProto)}}, }, nil, nil) @@ -779,11 +817,12 @@ func (s *Server) isMember(ctx context.Context, log *zap.Logger, chatID *commonpb // clientAllowedContent reports whether content is a message body a client may // author via SendMessage or EditMessage, and extracts the replied-to message ID -// when it is a reply. The permitted set is a whitelist — currently a plain text -// message, or a reply whose own content is text — so it excludes server-injected -// content (e.g. cash payment messages) and any content type added later until it is -// explicitly allowed. repliedMessageID is non-nil only for a valid reply, signaling -// the caller to verify the replied-to message exists and is repliable. +// when it is a reply. The permitted set is a whitelist — currently a text or media +// message, or a reply whose own body is text or media — so it excludes +// server-injected content (e.g. cash payment messages) and any content type added +// later until it is explicitly allowed. repliedMessageID is non-nil only for a +// valid reply, signaling the caller to verify the replied-to message exists and is +// repliable. func clientAllowedContent(content []*messagingpb.Content) (repliedMessageID *messagingpb.MessageId, ok bool) { if len(content) != 1 { return nil, false @@ -791,11 +830,13 @@ func clientAllowedContent(content []*messagingpb.Content) (repliedMessageID *mes switch c := content[0].Type.(type) { case *messagingpb.Content_Text: return nil, true + case *messagingpb.Content_Media: + return nil, validClientMedia(c.Media) case *messagingpb.Content_Reply: if len(c.Reply.Content) != 1 { return nil, false } - if _, isText := c.Reply.Content[0].Type.(*messagingpb.Content_Text); !isText { + if !validReplyBody(c.Reply.Content[0]) { return nil, false } return c.Reply.RepliedMessageId, true @@ -804,6 +845,19 @@ func clientAllowedContent(content []*messagingpb.Content) (repliedMessageID *mes } } +// validReplyBody reports whether a reply's body is content a client may author: +// text or well-formed media. +func validReplyBody(content *messagingpb.Content) bool { + switch c := content.Type.(type) { + case *messagingpb.Content_Text: + return true + case *messagingpb.Content_Media: + return validClientMedia(c.Media) + default: + return false + } +} + func (s *Server) messageExists(ctx context.Context, log *zap.Logger, chatID *commonpb.ChatId, messageID *messagingpb.MessageId) (bool, error) { exists, err := s.messages.MessageExists(ctx, chatID, messageID) if err != nil { diff --git a/messaging/tests/server.go b/messaging/tests/server.go index 1a1b549..f85c035 100644 --- a/messaging/tests/server.go +++ b/messaging/tests/server.go @@ -12,6 +12,7 @@ import ( "go.uber.org/zap/zaptest" "google.golang.org/grpc" + blobpb "github.com/code-payments/flipcash2-protobuf-api/generated/go/blob/v1" chatpb "github.com/code-payments/flipcash2-protobuf-api/generated/go/chat/v1" commonpb "github.com/code-payments/flipcash2-protobuf-api/generated/go/common/v1" eventpb "github.com/code-payments/flipcash2-protobuf-api/generated/go/event/v1" @@ -19,6 +20,8 @@ import ( "github.com/code-payments/flipcash2-server/auth" "github.com/code-payments/flipcash2-server/badge" + "github.com/code-payments/flipcash2-server/blob" + blob_memory "github.com/code-payments/flipcash2-server/blob/memory" "github.com/code-payments/flipcash2-server/chat" "github.com/code-payments/flipcash2-server/event" "github.com/code-payments/flipcash2-server/messaging" @@ -38,6 +41,8 @@ func RunServerTests(t *testing.T, badges badge.Store, chats chat.Store, messages testServer_SendMessage_Idempotent, testServer_SendReply, testServer_SendMessage_DisallowedContent, + testServer_SendMedia, + testServer_ResolvesMediaOnRead, testServer_SendMessage_Broadcast, testServer_EditMessage, testServer_DeleteMessage, @@ -78,6 +83,9 @@ type serverEnv struct { keysA model.KeyPair userB *commonpb.UserId keysB model.KeyPair + + blobStore blob.Store + blobAccess blob.AccessStore } func newServerEnv(t *testing.T, badges badge.Store, chats chat.Store, messages messaging.Store, profiles profile.Store) *serverEnv { @@ -106,8 +114,14 @@ func newServerEnv(t *testing.T, badges badge.Store, chats chat.Store, messages m LastActivity: at(1), })) - sender := messaging.NewSender(log, badges, chats, messages, profiles, ocp_data.NewTestDataProvider(), push.NewNoOpPusher(), bus) - server := messaging.NewServer(log, authz, chats, messages, sender) + blobStore := blob_memory.NewInMemory() + blobAccess := blob_memory.NewInMemoryAccessStore() + env.blobStore = blobStore + env.blobAccess = blobAccess + media := blob.NewIntegration(blobStore, blob_memory.NewInMemoryStorage(), blobAccess) + + sender := messaging.NewSender(log, badges, chats, messages, profiles, media, ocp_data.NewTestDataProvider(), push.NewNoOpPusher(), bus) + server := messaging.NewServer(log, authz, chats, messages, media, sender) cc := testutil.RunGRPCServer(t, log, testutil.WithService(func(s *grpc.Server) { messagingpb.RegisterMessagingServer(s, server) })) @@ -456,6 +470,33 @@ func lastMessageID(resp *messagingpb.GetMessagesResponse) *messagingpb.MessageId return msgs[len(msgs)-1].MessageId } +// putReadyBlob seeds a READY original blob owned by owner into the blob store and +// returns its id, so a media message can reference a real, shareable blob. +// newBlobID returns a random, well-formed (16-byte) blob id. +func (e *serverEnv) putReadyBlob(owner *commonpb.UserId) *blobpb.BlobId { + id := blob.MustGenerateID() + require.NoError(e.t, e.blobStore.CreatePending(e.ctx, &blob.Blob{ + ID: id, + Rendition: blob.RenditionOriginal, + Owner: owner, + State: blob.StatePending, + StorageKey: "images/x/original.png", + MimeType: "image/png", + SizeBytes: 1, + })) + _, err := e.blobStore.Advance(e.ctx, id, blob.StateReady, nil) + require.NoError(e.t, err) + return id +} + +// chatGrantedRead reports whether the chat has been granted read access to +// blobID — i.e. whether a chat-read grant was written when the blob was shared. +func (e *serverEnv) chatGrantedRead(blobID *blobpb.BlobId) bool { + has, err := e.blobAccess.HasGrant(e.ctx, blobID, blob.PrincipalForChat(e.chatID), blob.PermissionRead) + require.NoError(e.t, err) + return has +} + // ============================================================================ // Messaging // ============================================================================ @@ -480,6 +521,96 @@ func testServer_SendAndGet(t *testing.T, chats chat.Store, messages messaging.St require.Len(t, listResp.Messages.Messages, 1) } +func testServer_SendMedia(t *testing.T, chats chat.Store, messages messaging.Store, profiles profile.Store, badges badge.Store) { + e := newServerEnv(t, badges, chats, messages, profiles) + + // Media owned by the sender is sent, and the blob is granted to the chat. + ownedBlob := e.putReadyBlob(e.userA) + ownedResp, err := e.sendContent(e.keysA, mediaContent(ownedBlob), generateClientID()) + require.NoError(t, err) + require.Equal(t, messagingpb.SendMessageResponse_OK, ownedResp.Result) + require.True(t, e.chatGrantedRead(ownedBlob)) + + // Media owned by another user is denied, and nothing is granted. + othersBlob := e.putReadyBlob(e.userB) + deniedResp, err := e.sendContent(e.keysA, mediaContent(othersBlob), generateClientID()) + require.NoError(t, err) + require.Equal(t, messagingpb.SendMessageResponse_DENIED, deniedResp.Result) + require.False(t, e.chatGrantedRead(othersBlob)) + + // A reply whose body is media shares the media too. + original, err := e.send(e.keysA, "original", generateClientID()) + require.NoError(t, err) + replyBlob := e.putReadyBlob(e.userB) + replyResp, err := e.sendContent(e.keysB, replyMediaContent(original.Message.MessageId.Value, replyBlob), generateClientID()) + require.NoError(t, err) + require.Equal(t, messagingpb.SendMessageResponse_OK, replyResp.Result) + require.True(t, e.chatGrantedRead(replyBlob)) +} + +func testServer_ResolvesMediaOnRead(t *testing.T, chats chat.Store, messages messaging.Store, profiles profile.Store, badges badge.Store) { + e := newServerEnv(t, badges, chats, messages, profiles) + + // A sends a media message referencing a blob. + blobID := e.putReadyBlob(e.userA) + sendResp, err := e.sendContent(e.keysA, mediaContent(blobID), generateClientID()) + require.NoError(t, err) + require.Equal(t, messagingpb.SendMessageResponse_OK, sendResp.Result) + + // The send response itself carries the resolved metadata. + sent := sendResp.Message.Content[0].GetMedia().Items[0].Renditions[0] + require.NotNil(t, sent.Blob) + require.NotEmpty(t, sent.Blob.GetDownloadUrl().GetUrl()) + + // The live broadcast (message_sent event) carries the resolved metadata too. + e.waitForChatUpdate(e.userB, func(u *eventpb.ChatUpdate) bool { + if u.Events == nil || len(u.Events.Events) != 1 || len(u.Events.Events[0].Mutations) != 1 { + return false + } + sent := u.Events.Events[0].Mutations[0].GetMessageSent() + if sent == nil || sent.MessageId.Value != sendResp.Message.MessageId.Value { + return false + } + r := sent.Content[0].GetMedia().Items[0].Renditions[0] + return r.GetBlob().GetDownloadUrl().GetUrl() != "" + }) + + // B (a member) reads it, and the server fills the rendition's resolved metadata + // — a download URL it can fetch the bytes from — alongside the blob id. + getResp, err := e.getMessage(e.keysB, sendResp.Message.MessageId) + require.NoError(t, err) + require.Equal(t, messagingpb.GetMessageResponse_OK, getResp.Result) + + got := getResp.Message.Content[0].GetMedia().Items[0].Renditions[0] + require.Equal(t, blobID.Value, got.BlobId.Value) + require.NotNil(t, got.Blob) + require.Equal(t, "image/png", got.Blob.MimeType) + require.NotNil(t, got.Blob.DownloadUrl) + require.NotEmpty(t, got.Blob.DownloadUrl.Url) + + // The same metadata is filled on the list (GetMessages) path. + listResp, err := e.getMessagesByOptions(e.keysB, &commonpb.QueryOptions{}) + require.NoError(t, err) + require.Equal(t, messagingpb.GetMessagesResponse_OK, listResp.Result) + listed := listResp.Messages.Messages[0].Content[0].GetMedia().Items[0].Renditions[0] + require.NotNil(t, listed.Blob) + require.NotEmpty(t, listed.Blob.DownloadUrl.Url) + + // And on the delta (GetDelta) catch-up path. + deltaResps, err := e.getDelta(e.keysB, 0) + require.NoError(t, err) + deltaMsgs, _, _ := collectDelta(deltaResps) + var delivered *messagingpb.MediaItemRendition + for _, m := range deltaMsgs { + if m.MessageId.Value == sendResp.Message.MessageId.Value { + delivered = m.Content[0].GetMedia().Items[0].Renditions[0] + } + } + require.NotNil(t, delivered) + require.NotNil(t, delivered.Blob) + require.NotEmpty(t, delivered.Blob.DownloadUrl.Url) +} + func testServer_SendMessage_Idempotent(t *testing.T, chats chat.Store, messages messaging.Store, profiles profile.Store, badges badge.Store) { e := newServerEnv(t, badges, chats, messages, profiles) @@ -547,9 +678,30 @@ func testServer_SendReply(t *testing.T, chats chat.Store, messages messaging.Sto func testServer_SendMessage_DisallowedContent(t *testing.T, chats chat.Store, messages messaging.Store, profiles profile.Store, badges badge.Store) { e := newServerEnv(t, badges, chats, messages, profiles) + // A server-injected system message may not be authored by a client. systemResp, err := e.sendContent(e.keysA, systemContent("i joined"), generateClientID()) require.NoError(t, err) require.Equal(t, messagingpb.SendMessageResponse_DENIED, systemResp.Result) + + // Media referencing a non-ORIGINAL rendition is rejected: a client references + // only the original; the server derives and serves the rest. + derivedRole := mediaContent(blob.MustGenerateID()) + derivedRole[0].GetMedia().Items[0].Renditions[0].Role = messagingpb.MediaItemRendition_DISPLAY + roleResp, err := e.sendContent(e.keysA, derivedRole, generateClientID()) + require.NoError(t, err) + require.Equal(t, messagingpb.SendMessageResponse_DENIED, roleResp.Result) + + // Media whose item carries more than the single ORIGINAL rendition a client may + // supply is rejected. + extraRendition := mediaContent(blob.MustGenerateID()) + item := extraRendition[0].GetMedia().Items[0] + item.Renditions = append(item.Renditions, &messagingpb.MediaItemRendition{ + Role: messagingpb.MediaItemRendition_ORIGINAL, + BlobId: blob.MustGenerateID(), + }) + extraResp, err := e.sendContent(e.keysA, extraRendition, generateClientID()) + require.NoError(t, err) + require.Equal(t, messagingpb.SendMessageResponse_DENIED, extraResp.Result) } func testServer_SendMessage_Broadcast(t *testing.T, chats chat.Store, messages messaging.Store, profiles profile.Store, badges badge.Store) { @@ -655,6 +807,25 @@ func testServer_EditMessage(t *testing.T, chats chat.Store, messages messaging.S require.Equal(t, "edited again", reEdit.Message.Content[0].GetText().Text) require.Greater(t, reEdit.Message.EventSequence, editedSeq) + // Editing to media the sender owns shares the blob into the chat and resolves + // its metadata onto the returned message, just like a send. + ownedBlob := e.putReadyBlob(e.userA) + toMedia, err := e.editMessage(e.keysA, msgID, mediaContent(ownedBlob), reEdit.Message.EventSequence) + require.NoError(t, err) + require.Equal(t, messagingpb.EditMessageResponse_OK, toMedia.Result) + require.True(t, e.chatGrantedRead(ownedBlob)) + rendition := toMedia.Message.Content[0].GetMedia().Items[0].Renditions[0] + require.Equal(t, ownedBlob.Value, rendition.BlobId.Value) + require.NotNil(t, rendition.Blob) + require.NotEmpty(t, rendition.Blob.GetDownloadUrl().GetUrl()) + + // Editing to media owned by another user is denied, and nothing is granted. + othersBlob := e.putReadyBlob(e.userB) + deniedMedia, err := e.editMessage(e.keysA, msgID, mediaContent(othersBlob), toMedia.Message.EventSequence) + require.NoError(t, err) + require.Equal(t, messagingpb.EditMessageResponse_DENIED, deniedMedia.Result) + require.False(t, e.chatGrantedRead(othersBlob)) + // A non-editable message the caller authored (seeded directly as a system // message, which SendMessage won't accept) is rejected with CANNOT_EDIT. sysMsg, _, err := messages.PutMessage(e.ctx, e.chatID, e.userA, systemContent("system"), at(50), generateClientID(), false) @@ -664,7 +835,7 @@ func testServer_EditMessage(t *testing.T, chats chat.Store, messages messaging.S require.Equal(t, messagingpb.EditMessageResponse_CANNOT_EDIT, cannot.Result) // A tombstone is not editable: deleting then editing is rejected with CANNOT_EDIT. - del, err := e.deleteMessage(e.keysA, msgID, reEdit.Message.EventSequence) + del, err := e.deleteMessage(e.keysA, msgID, toMedia.Message.EventSequence) require.NoError(t, err) require.Equal(t, messagingpb.DeleteMessageResponse_OK, del.Result) cannotEditDeleted, err := e.editMessage(e.keysA, msgID, textContent("resurrect"), del.Message.EventSequence) diff --git a/messaging/tests/store.go b/messaging/tests/store.go index 0a8a5f9..97848e8 100644 --- a/messaging/tests/store.go +++ b/messaging/tests/store.go @@ -12,6 +12,7 @@ import ( "github.com/stretchr/testify/require" + blobpb "github.com/code-payments/flipcash2-protobuf-api/generated/go/blob/v1" commonpb "github.com/code-payments/flipcash2-protobuf-api/generated/go/common/v1" messagingpb "github.com/code-payments/flipcash2-protobuf-api/generated/go/messaging/v1" @@ -1240,6 +1241,34 @@ func replyContent(repliedMessageID uint64, text string) []*messagingpb.Content { }} } +// mediaContent builds a one-item media message referencing blobID as its ORIGINAL +// rendition — the shape a client sends. +func mediaContent(blobID *blobpb.BlobId) []*messagingpb.Content { + return []*messagingpb.Content{{ + Type: &messagingpb.Content_Media{Media: &messagingpb.MediaContent{ + Items: []*messagingpb.MediaItem{{ + Renditions: []*messagingpb.MediaItemRendition{{ + Role: messagingpb.MediaItemRendition_ORIGINAL, + BlobId: blobID, + }}, + }}, + }}, + }} +} + +// replyMediaContent builds a reply to repliedMessageID whose body is a media +// message referencing blobID. +func replyMediaContent(repliedMessageID uint64, blobID *blobpb.BlobId) []*messagingpb.Content { + return []*messagingpb.Content{{ + Type: &messagingpb.Content_Reply{ + Reply: &messagingpb.ReplyContent{ + RepliedMessageId: &messagingpb.MessageId{Value: repliedMessageID}, + Content: mediaContent(blobID), + }, + }, + }} +} + func messageText(m *messaging.Message) string { return m.Content[0].GetText().Text } diff --git a/task/chat_test.go b/task/chat_test.go index dca6821..7ba0bd6 100644 --- a/task/chat_test.go +++ b/task/chat_test.go @@ -18,6 +18,8 @@ import ( accountmemory "github.com/code-payments/flipcash2-server/account/memory" badgememory "github.com/code-payments/flipcash2-server/badge/memory" + "github.com/code-payments/flipcash2-server/blob" + blobmemory "github.com/code-payments/flipcash2-server/blob/memory" "github.com/code-payments/flipcash2-server/chat" chatmemory "github.com/code-payments/flipcash2-server/chat/memory" "github.com/code-payments/flipcash2-server/event" @@ -44,7 +46,8 @@ func TestExecutor_SendContactDmPaymentMessage(t *testing.T) { ocpData := ocp_data.NewTestDataProvider() bus := event.NewBus[*commonpb.UserId, *eventpb.Event]() - sender := messaging.NewSender(log, badges, chats, messages, profiles, ocpData, push.NewNoOpPusher(), bus) + media := blob.NewIntegration(blobmemory.NewInMemory(), blobmemory.NewInMemoryStorage(), blobmemory.NewInMemoryAccessStore()) + sender := messaging.NewSender(log, badges, chats, messages, profiles, media, ocpData, push.NewNoOpPusher(), bus) executor := task.NewExecutor(accounts, chats, sender, ocpData) integration := intent.NewIntegration(accounts, profiles)