Media upload and message support

blob/helpers.go

@@ -0,0 +1,65 @@
+package blob
+
+import (
+	"errors"
+
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/timestamppb"
+
+	blobpb "github.com/code-payments/flipchat-protobuf-api/generated/go/blob/v1"
+	commonpb "github.com/code-payments/flipchat-protobuf-api/generated/go/common/v1"
+
+	"github.com/code-payments/flipchat-server/model"
+)
+
+// Generates a unique BlobId
+func GenerateBlobID() *commonpb.BlobId {
+	return model.MustGenerateBlobID()
+}
+
+// Converts protobuf BlobType to internal BlobType
+func FromProtoBlobType(protoType blobpb.BlobType) (BlobType, error) {
+	switch protoType {
+	case blobpb.BlobType_BLOB_TYPE_IMAGE:
+		return BlobTypeImage, nil
+	case blobpb.BlobType_BLOB_TYPE_VIDEO:
+		return BlobTypeVideo, nil
+	case blobpb.BlobType_BLOB_TYPE_AUDIO:
+		return BlobTypeAudio, nil
+	default:
+		return BlobTypeUnknown, errors.New("unknown blob type")
+	}
+}
+
+// toProtoBlobType converts internal BlobType to protobuf BlobType
+func ToProtoBlobType(internalType BlobType) blobpb.BlobType {
+	switch internalType {
+	case BlobTypeImage:
+		return blobpb.BlobType_BLOB_TYPE_IMAGE
+	case BlobTypeVideo:
+		return blobpb.BlobType_BLOB_TYPE_VIDEO
+	case BlobTypeAudio:
+		return blobpb.BlobType_BLOB_TYPE_AUDIO
+	default:
+		return blobpb.BlobType_BLOB_TYPE_UNKNOWN
+	}
+}
+
+// Converts internal Blob to protobuf Blob, including unmarshaling metadata.
+func ToProtoBlob(blob *Blob) (*blobpb.Blob, error) {
+	var metadata blobpb.Blob_Metadata
+	err := proto.Unmarshal(blob.Metadata, &metadata)
+	if err != nil {
+		return nil, err
+	}
+
+	protoBlob := &blobpb.Blob{
+		BlobId:    blob.ID,
+		BlobType:  ToProtoBlobType(blob.Type),
+		OwnerId:   blob.UserID,
+		S3Url:     blob.S3URL,
+		Metadata:  &metadata,
+		CreatedAt: timestamppb.New(blob.CreatedAt),
+	}
+	return protoBlob, nil
+}

blob/memory/server_test.go

@@ -0,0 +1,21 @@
+package memory
+
+import (
+	"testing"
+
+	"github.com/code-payments/flipchat-server/blob/tests"
+
+	account "github.com/code-payments/flipchat-server/account/memory"
+	s3 "github.com/code-payments/flipchat-server/s3/memory"
+)
+
+func TestBlob_MemoryServer(t *testing.T) {
+	testAccountStore := account.NewInMemory()
+	testS3Store := s3.NewInMemory()
+	testStore := NewInMemory()
+
+	teardown := func() {
+		testStore.(*store).reset()
+	}
+	tests.RunBlobServerTests(t, testAccountStore, testStore, testS3Store, teardown)
+}

blob/memory/store.go

@@ -0,0 +1,53 @@
+package memory
+
+import (
+	"context"
+	"sync"
+
+	commonpb "github.com/code-payments/flipchat-protobuf-api/generated/go/common/v1"
+
+	"github.com/code-payments/flipchat-server/blob"
+	"github.com/code-payments/flipchat-server/model"
+)
+
+type store struct {
+	mu   sync.RWMutex
+	data map[string]*blob.Blob
+}
+
+func NewInMemory() blob.Store {
+	return &store{
+		data: make(map[string]*blob.Blob),
+	}
+}
+
+func (s *store) reset() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	s.data = make(map[string]*blob.Blob)
+}
+
+func (s *store) CreateBlob(ctx context.Context, b *blob.Blob) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	key := model.BlobIDString(b.ID)
+	if _, found := s.data[key]; found {
+		return blob.ErrExists
+	}
+	s.data[key] = b.Clone()
+	return nil
+}
+
+func (s *store) GetBlob(ctx context.Context, id *commonpb.BlobId) (*blob.Blob, error) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	key := model.BlobIDString(id)
+	bl, found := s.data[key]
+	if !found {
+		return nil, blob.ErrNotFound
+	}
+	return bl.Clone(), nil
+}

blob/memory/store_test.go

@@ -0,0 +1,15 @@
+package memory
+
+import (
+	"testing"
+
+	"github.com/code-payments/flipchat-server/blob/tests"
+)
+
+func TestBlob_MemoryStore(t *testing.T) {
+	testStore := NewInMemory()
+	teardown := func() {
+		// testStore.(*memory).reset()
+	}
+	tests.RunStoreTests(t, testStore, teardown)
+}

blob/postgres/common_test.go

@@ -0,0 +1,34 @@
+//go:build integration
+
+package postgres
+
+import (
+	"os"
+	"testing"
+
+	"github.com/sirupsen/logrus"
+
+	prismatest "github.com/code-payments/flipchat-server/database/prisma/test"
+
+	_ "github.com/jackc/pgx/v4/stdlib"
+)
+
+var testEnv *prismatest.TestEnv
+
+func TestMain(m *testing.M) {
+	log := logrus.StandardLogger()
+
+	// Create a new test environment
+	env, err := prismatest.NewTestEnv()
+	if err != nil {
+		log.WithError(err).Error("Error creating test environment")
+		os.Exit(1)
+	}
+
+	// Set the test environment
+	testEnv = env
+
+	// Run tests
+	code := m.Run()
+	os.Exit(code)
+}

blob/postgres/server_test.go

@@ -0,0 +1,30 @@
+//go:build integration
+
+package postgres
+
+import (
+	"testing"
+
+	account "github.com/code-payments/flipchat-server/account/postgres"
+	prismatest "github.com/code-payments/flipchat-server/database/prisma/test"
+	s3 "github.com/code-payments/flipchat-server/s3/memory"
+
+	"github.com/code-payments/flipchat-server/blob/tests"
+
+	_ "github.com/jackc/pgx/v4/stdlib"
+)
+
+func TestMessaging_PostgresServer(t *testing.T) {
+	client, disconnect := prismatest.NewTestClient(testEnv.DatabaseUrl, t)
+	defer disconnect()
+
+	accounts := account.NewInPostgres(client)
+	blobs := NewInPostgres(client)
+	s3s := s3.NewInMemory()
+
+	teardown := func() {
+		blobs.(*store).reset()
+	}
+
+	tests.RunBlobServerTests(t, accounts, blobs, s3s, teardown)
+}

blob/postgres/store.go

@@ -0,0 +1,86 @@
+package postgres
+
+import (
+	"context"
+	"fmt"
+
+	commonpb "github.com/code-payments/flipchat-protobuf-api/generated/go/common/v1"
+	pg "github.com/code-payments/flipchat-server/database/postgres"
+
+	"github.com/code-payments/flipchat-server/blob"
+	"github.com/code-payments/flipchat-server/database/prisma/db"
+)
+
+type store struct {
+	client *db.PrismaClient
+}
+
+func NewInPostgres(client *db.PrismaClient) blob.Store {
+	return &store{
+		client,
+	}
+}
+
+func (s *store) reset() {
+	ctx := context.Background()
+
+	blobs := s.client.Blob.FindMany().Delete().Tx()
+	err := s.client.Prisma.Transaction(blobs).Exec(ctx)
+	if err != nil {
+		panic(err)
+	}
+}
+
+func (s *store) CreateBlob(ctx context.Context, b *blob.Blob) error {
+	found, err := s.client.Blob.FindUnique(
+		db.Blob.ID.Equals(b.ID.Value),
+	).Exec(ctx)
+
+	if err == nil && found != nil {
+		return blob.ErrExists
+	}
+
+	encodedUserId := pg.Encode(b.UserID.Value)
+
+	_, err = s.client.Blob.CreateOne(
+		db.Blob.ID.Set(b.ID.Value),
+		db.Blob.UserID.Set(encodedUserId),
+		db.Blob.Type.Set(int(b.Type)),
+		db.Blob.S3URL.Set(b.S3URL),
+		db.Blob.Size.Set(int(b.Size)),
+		db.Blob.Metadata.Set(b.Metadata),
+		db.Blob.Flagged.Set(b.Flagged),
+		db.Blob.CreatedAt.Set(b.CreatedAt),
+	).Exec(ctx)
+
+	if err != nil {
+		return fmt.Errorf("failed to create blob: %w", err)
+	}
+	return nil
+}
+
+func (s *store) GetBlob(ctx context.Context, id *commonpb.BlobId) (*blob.Blob, error) {
+	row, err := s.client.Blob.FindUnique(
+		db.Blob.ID.Equals(id.Value),
+	).Exec(ctx)
+
+	if err != nil || row == nil {
+		return nil, blob.ErrNotFound
+	}
+
+	userID, err := pg.Decode(row.UserID)
+	if err != nil {
+		return nil, err
+	}
+
+	return &blob.Blob{
+		ID:        id,
+		UserID:    &commonpb.UserId{Value: userID},
+		Type:      blob.BlobType(row.Type),
+		S3URL:     row.S3URL,
+		Size:      int64(row.Size),
+		Metadata:  row.Metadata,
+		Flagged:   row.Flagged,
+		CreatedAt: row.CreatedAt,
+	}, nil
+}

blob/postgres/store_test.go

@@ -0,0 +1,24 @@
+//go:build integration
+
+package postgres
+
+import (
+	"testing"
+
+	prismatest "github.com/code-payments/flipchat-server/database/prisma/test"
+
+	"github.com/code-payments/flipchat-server/blob/tests"
+
+	_ "github.com/jackc/pgx/v4/stdlib"
+)
+
+func TestBlob_PostgresStore(t *testing.T) {
+	client, disconnect := prismatest.NewTestClient(testEnv.DatabaseUrl, t)
+	defer disconnect()
+
+	testStore := NewInPostgres(client)
+	teardown := func() {
+		testStore.(*store).reset()
+	}
+	tests.RunStoreTests(t, testStore, teardown)
+}