FlowyLabs/llink
1
0
Fork 0
Code Issues 71 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

refactor: update api and client to reference humanIds

Browse Source
This commit is contained in:
talksik
2026-03-24 19:08:51 -07:00
parent 3bf3f16be7
commit 57a1cbc696
24 changed files with 666 additions and 416 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go/internal/particle/service.go
+3 -42
View File
@@ -4,7 +4,6 @@ import (
"context"
"encoding/json"
"errors"
"log/slog"
"time"
"github.com/flowy-live/llink/internal/network"
@@ -14,6 +13,7 @@ import (
const defaultPageSize = 50
// NOTE: this service is deprecated as we use firestore for particle data
type Service interface {
// Create creates a new particle. Caller must be a network member (verified by handler).
// Returns ErrInvalidType, ErrInvalidData, ErrMembersRequired, ErrInvalidParent, ErrAccessDenied, or ErrCapacityExceeded.
@@ -198,26 +198,11 @@ func (s *serviceImpl) Create(ctx context.Context, input CreateInput, requesterEm
return nil, err
}
p.Data = data
// Check and increment capacity
err = s.networkSvc.IncrementOpenStreamCount(ctx, input.NetworkID)
if err != nil {
if errors.Is(err, network.ErrCapacityExceeded) {
return nil, ErrCapacityExceeded
}
return nil, err
}
}
// Create the particle
created, err := s.repo.create(ctx, p)
if err != nil {
// If we incremented the stream count but creation failed, decrement it
if input.Type == TypeStream {
if decErr := s.networkSvc.DecrementOpenStreamCount(ctx, input.NetworkID); decErr != nil {
slog.Warn("failed to rollback stream count after particle creation failure", "error", decErr, "network_id", input.NetworkID)
}
}
return nil, err
}
@@ -334,7 +319,7 @@ func (s *serviceImpl) Delete(ctx context.Context, id, requesterEmail string) err
}
// Get the particle to check if it's an open stream
p, err := s.repo.getByID(ctx, id)
_, err = s.repo.getByID(ctx, id)
if err != nil {
if errors.Is(err, errNotFound) {
return ErrNotFound
@@ -342,13 +327,6 @@ func (s *serviceImpl) Delete(ctx context.Context, id, requesterEmail string) err
return err
}
// If it's an open stream, decrement the count
if p.Type == TypeStream && getStreamStatus(p.Data) == string(StreamStatusOpen) {
if err := s.networkSvc.DecrementOpenStreamCount(ctx, p.NetworkID); err != nil {
return err
}
}
err = s.repo.delete(ctx, id)
if errors.Is(err, errNotFound) {
return ErrNotFound
@@ -454,30 +432,14 @@ func (s *serviceImpl) OpenStream(ctx context.Context, id, requesterEmail string)
return ErrStreamAlreadyOpen
}
// Check and increment capacity
err = s.networkSvc.IncrementOpenStreamCount(ctx, p.NetworkID)
if err != nil {
if errors.Is(err, network.ErrCapacityExceeded) {
return ErrCapacityExceeded
}
return err
}
// Update stream status in data
newData, err := setStreamStatus(p.Data, string(StreamStatusOpen))
if err != nil {
if decErr := s.networkSvc.DecrementOpenStreamCount(ctx, p.NetworkID); decErr != nil {
slog.Warn("failed to rollback stream count after status update failure", "error", decErr, "network_id", p.NetworkID, "particle_id", id)
}
return err
}
err = s.repo.update(ctx, id, newData, time.Now())
if err != nil {
// Rollback the capacity increment
if decErr := s.networkSvc.DecrementOpenStreamCount(ctx, p.NetworkID); decErr != nil {
slog.Warn("failed to rollback stream count after particle update failure", "error", decErr, "network_id", p.NetworkID, "particle_id", id)
}
if errors.Is(err, errNotFound) {
return ErrNotFound
}
@@ -536,8 +498,7 @@ func (s *serviceImpl) CloseStream(ctx context.Context, id, requesterEmail string
return err
}
// Decrement capacity
return s.networkSvc.DecrementOpenStreamCount(ctx, p.NetworkID)
return nil
}
func (s *serviceImpl) SetVisibility(ctx context.Context, id string, mode VisibilityMode, requesterEmail string) error {
go/internal/particle/service_test.go
-52
View File
@@ -70,58 +70,6 @@ func TestParticleService_CreateAndGet(t *testing.T) {
// Service assumes caller is already verified as network member
}
func TestParticleService_StreamCapacity(t *testing.T) {
ctx := context.Background()
networkSvc := network.NewService(dbPool)
svc := particle.NewService(dbPool, networkSvc)
// Create a network with capacity 2
net, err := networkSvc.Create(ctx, "Capacity Test Network", "[email protected]")
assert.NoError(t, err)
err = networkSvc.SetOpenStreamCapacity(ctx, net.ID, 2)
assert.NoError(t, err)
// Create first stream - should succeed
input := particle.CreateInput{
Type: particle.TypeStream,
NetworkID: net.ID,
Data: json.RawMessage(`{"name":"Stream 1","status":"open"}`),
}
stream1, err := svc.Create(ctx, input, "[email protected]")
assert.NoError(t, err)
// Create second stream - should succeed
input.Data = json.RawMessage(`{"name":"Stream 2","status":"open"}`)
stream2, err := svc.Create(ctx, input, "[email protected]")
assert.NoError(t, err)
// Create third stream - should fail with capacity exceeded
input.Data = json.RawMessage(`{"name":"Stream 3","status":"open"}`)
_, err = svc.Create(ctx, input, "[email protected]")
assert.Error(t, err)
assert.ErrorIs(t, err, particle.ErrCapacityExceeded)
// Close a stream
err = svc.CloseStream(ctx, stream1.ID, "[email protected]")
assert.NoError(t, err)
// Now we can create another stream
stream3, err := svc.Create(ctx, input, "[email protected]")
assert.NoError(t, err)
assert.NotEmpty(t, stream3.ID)
// Verify stream2 is still open
found, err := svc.GetByID(ctx, stream2.ID, "[email protected]")
assert.NoError(t, err)
assert.Equal(t, string(particle.StreamStatusOpen), getStreamStatus(found.Data))
// Verify stream1 is closed
found, err = svc.GetByID(ctx, stream1.ID, "[email protected]")
assert.NoError(t, err)
assert.Equal(t, string(particle.StreamStatusClosed), getStreamStatus(found.Data))
}
func TestParticleService_NestedParticles(t *testing.T) {
ctx := context.Background()
networkSvc := network.NewService(dbPool)