root
44406b72db
Moteva-style AI design workbench replica. Home prompt creates a project; the project page provides an infinite canvas with node drag, zoom/pan, a design chat panel, history replay, image asset upload, and project save/regenerate. - Backend: go-zero, DDD layering, sqlc/pgx, optional PostgreSQL; memory or Redis cache; asynq job queue; MinIO/S3/R2/OSS object storage; sky-valley/pi agent runtime adapter. - Frontend: Next.js App Router + Vite artifact build, TypeScript, i18n, shadcn/ui components, auth (OTP/Turnstile/Google/WeChat). - Deploy: Docker Compose and k3s manifests. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
129 lines
2.4 KiB
Go
129 lines
2.4 KiB
Go
package realtime
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
|
|
"github.com/redis/go-redis/v9"
|
|
)
|
|
|
|
type RedisOptions struct {
|
|
Addr string
|
|
Password string
|
|
DB int
|
|
Prefix string
|
|
}
|
|
|
|
type RedisBus struct {
|
|
client *redis.Client
|
|
prefix string
|
|
}
|
|
|
|
type redisSubscription struct {
|
|
pubsub *redis.PubSub
|
|
ch chan Event
|
|
closeOnce sync.Once
|
|
chOnce sync.Once
|
|
}
|
|
|
|
func NewRedisBus(ctx context.Context, opts RedisOptions) (*RedisBus, error) {
|
|
opts.Addr = strings.TrimSpace(opts.Addr)
|
|
if opts.Addr == "" {
|
|
opts.Addr = "localhost:6379"
|
|
}
|
|
prefix := strings.TrimSpace(opts.Prefix)
|
|
if prefix == "" {
|
|
prefix = "canvas:realtime"
|
|
}
|
|
client := redis.NewClient(&redis.Options{
|
|
Addr: opts.Addr,
|
|
Password: opts.Password,
|
|
DB: opts.DB,
|
|
})
|
|
if err := client.Ping(ctx).Err(); err != nil {
|
|
_ = client.Close()
|
|
return nil, err
|
|
}
|
|
return &RedisBus{client: client, prefix: prefix}, nil
|
|
}
|
|
|
|
func (b *RedisBus) Publish(ctx context.Context, event Event) error {
|
|
if b == nil || b.client == nil {
|
|
return nil
|
|
}
|
|
data, err := json.Marshal(event)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return b.client.Publish(ctx, b.channel(event.ProjectID), data).Err()
|
|
}
|
|
|
|
func (b *RedisBus) Subscribe(ctx context.Context, projectID string) (Subscription, error) {
|
|
if b == nil || b.client == nil {
|
|
return nil, nil
|
|
}
|
|
pubsub := b.client.Subscribe(ctx, b.channel(projectID))
|
|
if _, err := pubsub.Receive(ctx); err != nil {
|
|
_ = pubsub.Close()
|
|
return nil, err
|
|
}
|
|
sub := &redisSubscription{
|
|
pubsub: pubsub,
|
|
ch: make(chan Event, 32),
|
|
}
|
|
go sub.run(ctx)
|
|
return sub, nil
|
|
}
|
|
|
|
func (b *RedisBus) Close() error {
|
|
if b == nil || b.client == nil {
|
|
return nil
|
|
}
|
|
return b.client.Close()
|
|
}
|
|
|
|
func (b *RedisBus) channel(projectID string) string {
|
|
return fmt.Sprintf("%s:project:%s", b.prefix, strings.TrimSpace(projectID))
|
|
}
|
|
|
|
func (s *redisSubscription) C() <-chan Event {
|
|
return s.ch
|
|
}
|
|
|
|
func (s *redisSubscription) Close() error {
|
|
var err error
|
|
s.closeOnce.Do(func() {
|
|
err = s.pubsub.Close()
|
|
})
|
|
return err
|
|
}
|
|
|
|
func (s *redisSubscription) run(ctx context.Context) {
|
|
defer s.closeChannel()
|
|
defer s.Close()
|
|
for {
|
|
msg, err := s.pubsub.ReceiveMessage(ctx)
|
|
if err != nil {
|
|
return
|
|
}
|
|
var event Event
|
|
if err := json.Unmarshal([]byte(msg.Payload), &event); err != nil {
|
|
continue
|
|
}
|
|
select {
|
|
case s.ch <- event:
|
|
case <-ctx.Done():
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *redisSubscription) closeChannel() {
|
|
s.chOnce.Do(func() {
|
|
close(s.ch)
|
|
})
|
|
}
|