From f1a27996d398cc2422466c01ffb252219cd096fe Mon Sep 17 00:00:00 2001 From: A Date: Mon, 6 Jul 2026 14:31:19 +0800 Subject: [PATCH] feat: sync group call livestream support --- cmd/telesrv/main.go | 27 ++ .../migrations/0053_group_call_rtmp.down.sql | 2 + deploy/migrations/0053_group_call_rtmp.up.sql | 10 + .../0054_group_call_scheduled.down.sql | 2 + .../0054_group_call_scheduled.up.sql | 10 + .../0055_group_call_join_as.down.sql | 1 + .../migrations/0055_group_call_join_as.up.sql | 4 + go.mod | 7 + go.sum | 27 ++ internal/app/groupcalls/service.go | 71 ++++- internal/app/livestream/part.go | 70 +++++ internal/app/livestream/part_test.go | 103 +++++++ internal/app/livestream/rtmp.go | 68 +++++ internal/app/livestream/segmenter.go | 285 ++++++++++++++++++ internal/app/livestream/service.go | 188 ++++++++++++ internal/config/config.go | 20 ++ internal/domain/channel.go | 5 + internal/domain/groupcall.go | 29 +- internal/domain/livestream.go | 20 ++ internal/rpc/convert_channels_core.go | 5 + internal/rpc/convert_phone_group.go | 29 +- internal/rpc/deps.go | 20 +- internal/rpc/phone_group_call.go | 116 +++++-- internal/rpc/phone_group_call_joinas_test.go | 113 +++++++ internal/rpc/phone_group_call_rtmp.go | 192 ++++++++++++ internal/rpc/phone_group_call_rtmp_test.go | 267 ++++++++++++++++ internal/rpc/phone_group_call_scheduled.go | 112 +++++++ .../rpc/phone_group_call_scheduled_test.go | 151 ++++++++++ internal/rpc/phone_group_call_updates.go | 17 +- internal/rpc/phone_register.go | 3 + internal/rpc/phone_stubs.go | 22 +- internal/rpc/router.go | 6 + internal/rpc/sticker_link_smoke_test.go | 2 +- internal/rpc/upload.go | 60 +++- internal/store/groupcall.go | 11 + internal/store/memory/groupcall.go | 85 +++++- internal/store/postgres/groupcall.go | 142 +++++++-- 37 files changed, 2219 insertions(+), 83 deletions(-) create mode 100644 deploy/migrations/0053_group_call_rtmp.down.sql create mode 100644 deploy/migrations/0053_group_call_rtmp.up.sql create mode 100644 deploy/migrations/0054_group_call_scheduled.down.sql create mode 100644 deploy/migrations/0054_group_call_scheduled.up.sql create mode 100644 deploy/migrations/0055_group_call_join_as.down.sql create mode 100644 deploy/migrations/0055_group_call_join_as.up.sql create mode 100644 internal/app/livestream/part.go create mode 100644 internal/app/livestream/part_test.go create mode 100644 internal/app/livestream/rtmp.go create mode 100644 internal/app/livestream/segmenter.go create mode 100644 internal/app/livestream/service.go create mode 100644 internal/domain/livestream.go create mode 100644 internal/rpc/phone_group_call_joinas_test.go create mode 100644 internal/rpc/phone_group_call_rtmp.go create mode 100644 internal/rpc/phone_group_call_rtmp_test.go create mode 100644 internal/rpc/phone_group_call_scheduled.go create mode 100644 internal/rpc/phone_group_call_scheduled_test.go diff --git a/cmd/telesrv/main.go b/cmd/telesrv/main.go index a4a4558e..92a65131 100644 --- a/cmd/telesrv/main.go +++ b/cmd/telesrv/main.go @@ -35,6 +35,7 @@ import ( groupcallsapp "telesrv/internal/app/groupcalls" "telesrv/internal/app/help" "telesrv/internal/app/langpack" + "telesrv/internal/app/livestream" "telesrv/internal/app/maintenance" messageapp "telesrv/internal/app/messages" passkeyapp "telesrv/internal/app/passkey" @@ -206,6 +207,15 @@ func startDebugServer(ctx context.Context, addr string, logger *zap.Logger) { } // externalMediaOption 按配置启用外链媒体抓取；禁用时返回 nil（NewService 跳过 nil option）。 +// liveStreamDep 把可能为 nil 的 *livestream.Service 转成 rpc.LiveStreamsService， +// 避免 typed-nil interface（nil 具体指针装进接口后 != nil 的坑）。 +func liveStreamDep(s *livestream.Service) rpc.LiveStreamsService { + if s == nil { + return nil + } + return s +} + func externalMediaOption(cfg config.Config) filesapp.Option { if !cfg.ExternalMediaEnable { return nil @@ -493,6 +503,21 @@ func run(logger *zap.Logger) error { } sfuService = pionSFU } + // 频道 RTMP 直播媒体面（Live Stream）：内嵌 RTMP ingest（OBS 推流）+ ffmpeg + // 切段。未启用时信令仍可用，观众停留在"等待推流"占位。 + var liveStreamService *livestream.Service + if cfg.LiveStreamEnable { + liveStreamService = livestream.NewService(livestream.Config{ + ListenAddr: cfg.LiveStreamRtmpAddr, + FFmpegPath: cfg.LiveStreamFFmpegPath, + WorkDir: cfg.LiveStreamWorkDir, + SegmentKeep: cfg.LiveStreamSegmentKeep, + }, groupCallsService, logger.Named("livestream")) + if err := liveStreamService.Start(); err != nil { + return fmt.Errorf("init live stream: %w", err) + } + defer liveStreamService.Close() + } // 私聊通话中继（P3）：内嵌 TURN/STUN，phoneCall.connections 经 phoneConnectionWebrtc // 下发。未启用时退回 P1 的纯信令 LAN 直连。 turnService := turnsrv.Service(turnsrv.Disabled()) @@ -591,6 +616,7 @@ func run(logger *zap.Logger) error { CallSignalingMaxBytes: cfg.CallSignalingMaxBytes, CallForceRelay: cfg.CallForceRelay, GroupCallMaxParticipants: cfg.GroupCallMaxParticipants, + RtmpIngestURL: cfg.LiveStreamRtmpURL, // PFS temp→perm 解析缓存 5s：削减每帧 ResolveAuthKey 的 PG 查询。显式撤销会清缓存并 // 断开连接；re-bind 即时失效（onAuthBindTempAuthKey）。 TempKeyResolveCacheTTL: 5 * time.Second, @@ -619,6 +645,7 @@ func run(logger *zap.Logger) error { Passkey: passkeyService, Themes: themeService, GroupCalls: groupCallsService, + LiveStreams: liveStreamDep(liveStreamService), SFU: sfuService, TURN: turnService, LangPack: langPackService, diff --git a/deploy/migrations/0053_group_call_rtmp.down.sql b/deploy/migrations/0053_group_call_rtmp.down.sql new file mode 100644 index 00000000..53618a27 --- /dev/null +++ b/deploy/migrations/0053_group_call_rtmp.down.sql @@ -0,0 +1,2 @@ +DROP TABLE IF EXISTS group_call_rtmp_keys; +ALTER TABLE group_calls DROP COLUMN IF EXISTS rtmp_stream; diff --git a/deploy/migrations/0053_group_call_rtmp.up.sql b/deploy/migrations/0053_group_call_rtmp.up.sql new file mode 100644 index 00000000..09b20796 --- /dev/null +++ b/deploy/migrations/0053_group_call_rtmp.up.sql @@ -0,0 +1,10 @@ +-- Live Stream（RTMP 直播）：group_calls 增加 rtmp_stream 标记； +-- per-channel 持久 RTMP 推流密钥（revoke 轮换后旧 key 立即失效）。 +ALTER TABLE group_calls + ADD COLUMN IF NOT EXISTS rtmp_stream boolean DEFAULT false NOT NULL; + +CREATE TABLE IF NOT EXISTS group_call_rtmp_keys ( + channel_id bigint PRIMARY KEY, + stream_key text NOT NULL, + updated_at integer NOT NULL +); diff --git a/deploy/migrations/0054_group_call_scheduled.down.sql b/deploy/migrations/0054_group_call_scheduled.down.sql new file mode 100644 index 00000000..58eebde7 --- /dev/null +++ b/deploy/migrations/0054_group_call_scheduled.down.sql @@ -0,0 +1,2 @@ +DROP TABLE IF EXISTS group_call_schedule_subscribers; +ALTER TABLE group_calls DROP COLUMN IF EXISTS schedule_date; diff --git a/deploy/migrations/0054_group_call_scheduled.up.sql b/deploy/migrations/0054_group_call_scheduled.up.sql new file mode 100644 index 00000000..37c4dd6f --- /dev/null +++ b/deploy/migrations/0054_group_call_scheduled.up.sql @@ -0,0 +1,10 @@ +-- Scheduled video chat：group_calls 增加 schedule_date（>0=定时未开始）； +-- 开播提醒订阅（toggleGroupCallStartSubscription 的 per-user 状态）。 +ALTER TABLE group_calls + ADD COLUMN IF NOT EXISTS schedule_date integer DEFAULT 0 NOT NULL; + +CREATE TABLE IF NOT EXISTS group_call_schedule_subscribers ( + call_id bigint NOT NULL, + user_id bigint NOT NULL, + PRIMARY KEY (call_id, user_id) +); diff --git a/deploy/migrations/0055_group_call_join_as.down.sql b/deploy/migrations/0055_group_call_join_as.down.sql new file mode 100644 index 00000000..e3c4de24 --- /dev/null +++ b/deploy/migrations/0055_group_call_join_as.down.sql @@ -0,0 +1 @@ +ALTER TABLE group_call_participants DROP COLUMN IF EXISTS join_as_channel_id; diff --git a/deploy/migrations/0055_group_call_join_as.up.sql b/deploy/migrations/0055_group_call_join_as.up.sql new file mode 100644 index 00000000..a210d415 --- /dev/null +++ b/deploy/migrations/0055_group_call_join_as.up.sql @@ -0,0 +1,4 @@ +-- join_as：参与者可以以频道/群本身的身份入会（匿名管理员语义）。 +-- 0 = 以本人用户身份；唯一键仍是 (call_id, user_id)，换身份 rejoin 为替换。 +ALTER TABLE group_call_participants + ADD COLUMN IF NOT EXISTS join_as_channel_id bigint DEFAULT 0 NOT NULL; diff --git a/go.mod b/go.mod index 2a557cb7..f9ab166f 100644 --- a/go.mod +++ b/go.mod @@ -23,6 +23,7 @@ require ( github.com/pion/transport/v4 v4.0.2 github.com/pion/turn/v5 v5.0.10 github.com/redis/go-redis/v9 v9.20.0 + github.com/yutopp/go-rtmp v0.0.7 go.uber.org/multierr v1.11.0 go.uber.org/zap v1.28.0 golang.org/x/crypto v0.53.0 @@ -45,22 +46,28 @@ require ( github.com/google/uuid v1.6.0 // indirect github.com/gotd/log v0.1.0 // indirect github.com/gotd/neo v0.1.5 // indirect + github.com/hashicorp/errwrap v1.1.0 // indirect + github.com/hashicorp/go-multierror v1.1.0 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/puddle/v2 v2.2.2 // indirect github.com/klauspost/compress v1.19.0 // indirect github.com/mattn/go-colorable v0.1.14 // indirect github.com/mattn/go-isatty v0.0.22 // indirect + github.com/mitchellh/mapstructure v1.4.1 // indirect github.com/ogen-go/ogen v1.22.0 // indirect github.com/pion/mdns/v2 v2.1.0 // indirect github.com/pion/randutil v0.1.0 // indirect github.com/pion/stun/v3 v3.1.6 // indirect + github.com/pkg/errors v0.9.1 // indirect github.com/refraction-networking/utls v1.8.2 // indirect github.com/segmentio/asm v1.2.1 // indirect github.com/shopspring/decimal v1.4.0 // indirect + github.com/sirupsen/logrus v1.9.3 // indirect github.com/wlynxg/anet v0.0.5 // indirect github.com/x448/float16 v0.8.4 // indirect github.com/yuin/goldmark v1.8.2 // indirect + github.com/yutopp/go-amf0 v0.1.0 // indirect go.opentelemetry.io/otel v1.44.0 // indirect go.opentelemetry.io/otel/metric v1.44.0 // indirect go.opentelemetry.io/otel/trace v1.44.0 // indirect diff --git a/go.sum b/go.sum index 4296e7db..25db2642 100644 --- a/go.sum +++ b/go.sum @@ -19,6 +19,7 @@ github.com/containerd/errdefs v1.0.0/go.mod h1:+YBYIdtsnF4Iw6nWZhJcqGSg/dwvV7tyJ github.com/containerd/errdefs/pkg v0.3.0 h1:9IKJ06FvyNlexW690DXuQNx2KA2cUJXx151Xdx3ZPPE= github.com/containerd/errdefs/pkg v0.3.0/go.mod h1:NJw6s9HwNuRhnjJhM7pylWwMyAkmCQvQ4GpJHEqRLVk= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/dhui/dktest v0.4.6 h1:+DPKyScKSEp3VLtbMDHcUq6V5Lm5zfZZVb0Sk7Ahom4= @@ -37,6 +38,8 @@ github.com/fatih/color v1.19.0 h1:Zp3PiM21/9Ld6FzSKyL5c/BULoe/ONr9KlbYVOfG8+w= github.com/fatih/color v1.19.0/go.mod h1:zNk67I0ZUT1bEGsSGyCZYZNrHuTkJJB+r6Q9VuMi0LE= github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg= github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= +github.com/fortytw2/leaktest v1.2.0 h1:cj6GCiwJDH7l3tMHLjZDo0QqPtrXJiWSI9JgpeQKw+Q= +github.com/fortytw2/leaktest v1.2.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g= github.com/fxamacker/cbor/v2 v2.8.0 h1:fFtUGXUzXPHTIUdne5+zzMPTfffl3RD5qYnkY40vtxU= github.com/fxamacker/cbor/v2 v2.8.0/go.mod h1:vM4b+DJCtHn+zz7h3FFp/hDAI9WNWCsZj23V5ytsSxQ= github.com/ghodss/yaml v1.0.0 h1:wQHKEahhL6wmXdzwWG11gIVCkOv05bNOh+Rxn0yngAk= @@ -74,6 +77,11 @@ github.com/gotd/td v0.159.0 h1:kKXt2NLmfIOgebbFS34FSlZbdydaf5fsta+nP69nP+w= github.com/gotd/td v0.159.0/go.mod h1:rdZ2NfOMUViApJa3EvYJ94GAxENjCB0b98tJbfS9NCc= github.com/gotd/tl v0.4.0 h1:8k2z0drujiPyhpLDa9PRm/yU1Gwlfn3iUzeInPiXwMA= github.com/gotd/tl v0.4.0/go.mod h1:CMIcjPWFS4qxxJ+1Ce7U/ilbtPrkoVo/t8uhN5Y/D7c= +github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= +github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY2I= +github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= +github.com/hashicorp/go-multierror v1.1.0 h1:B9UzwGQJehnUY1yNrnwREHc3fGbC2xefo8g4TbElacI= +github.com/hashicorp/go-multierror v1.1.0/go.mod h1:spPvp8C1qA32ftKqdAHm4hHTbPw+vmowP0z+KUhOZdA= github.com/jackc/pgerrcode v0.0.0-20220416144525-469b46aa5efa h1:s+4MhCQ6YrzisK6hFJUX53drDT4UsSW3DEhKn0ifuHw= github.com/jackc/pgerrcode v0.0.0-20220416144525-469b46aa5efa/go.mod h1:a/s9Lp5W7n/DD0VrVoyJ00FbP2ytTPDVOivvn2bMlds= github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= @@ -98,6 +106,8 @@ github.com/mattn/go-colorable v0.1.14 h1:9A9LHSqF/7dyVVX6g0U9cwm9pG3kP9gSzcuIPHP github.com/mattn/go-colorable v0.1.14/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8= github.com/mattn/go-isatty v0.0.22 h1:j8l17JJ9i6VGPUFUYoTUKPSgKe/83EYU2zBC7YNKMw4= github.com/mattn/go-isatty v0.0.22/go.mod h1:ZXfXG4SQHsB/w3ZeOYbR0PrPwLy+n6xiMrJlRFqopa4= +github.com/mitchellh/mapstructure v1.4.1 h1:CpVNEelQCZBooIPDn+AR3NpivK/TIKU8bDxdASFVQag= +github.com/mitchellh/mapstructure v1.4.1/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo= github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0= github.com/moby/docker-image-spec v1.3.1/go.mod h1:eKmb5VW8vQEh/BAr2yvVNvuiJuY6UIocYsFu/DxxRpo= github.com/moby/term v0.5.0 h1:xt8Q1nalod/v7BqbG21f8mQPqH+xAaC9C3N3wfWbVP0= @@ -159,9 +169,18 @@ github.com/sergi/go-diff v1.1.0 h1:we8PVUC3FE2uYfodKH/nBHMSetSfHDR6scGdBi+erh0= github.com/sergi/go-diff v1.1.0/go.mod h1:STckp+ISIX8hZLjrqAeVduY0gWCT9IjLuqbuNXdaHfM= github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp81k= github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME= +github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0= +github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ= +github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= +github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= +github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= +github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/wlynxg/anet v0.0.5 h1:J3VJGi1gvo0JwZ/P1/Yc/8p63SoW98B5dHkYDmpgvvU= @@ -172,6 +191,11 @@ github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZ github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E= github.com/yuin/goldmark v1.8.2 h1:kEGpgqJXdgbkhcOgBxkC0X0PmoPG1ZyoZ117rDVp4zE= github.com/yuin/goldmark v1.8.2/go.mod h1:ip/1k0VRfGynBgxOz0yCqHrbZXhcjxyuS66Brc7iBKg= +github.com/yutopp/go-amf0 v0.1.0 h1:a3UeBZG7nRF0zfvmPn2iAfNo1RGzUpHz1VyJD2oGrik= +github.com/yutopp/go-amf0 v0.1.0/go.mod h1:QzDOBr9RV6sQh6E5GFEJROZbU0iQKijORBmprkb3FIk= +github.com/yutopp/go-flv v0.3.1/go.mod h1:pAlHPSVRMv5aCUKmGOS/dZn/ooTgnc09qOPmiUNMubs= +github.com/yutopp/go-rtmp v0.0.7 h1:sKKm1MVV3ANbJHZlf3Kq8ecq99y5U7XnDUDxSjuK7KU= +github.com/yutopp/go-rtmp v0.0.7/go.mod h1:KSwrC9Xj5Kf18EUlk1g7CScecjXfIqc0J5q+S0u6Irc= github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= @@ -207,6 +231,9 @@ golang.org/x/net v0.56.0/go.mod h1:D3Ku6r+V6JROoZK144D2XfMHFcMq/0zSfLelVTCFKec= golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM= golang.org/x/sync v0.21.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sys v0.0.0-20191026070338-33540a1f6037/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.3.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.46.0 h1:noSf2Fq6F8DBgS+LysIkx7rIExoNHJsxOAtPp4rthXw= golang.org/x/sys v0.46.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/text v0.38.0 h1:sXmwo9DwP3OK9EZ7PqAdaooSGozfl/3a6/xJcbzPRhE= diff --git a/internal/app/groupcalls/service.go b/internal/app/groupcalls/service.go index 6de66e5e..6007b8fc 100644 --- a/internal/app/groupcalls/service.go +++ b/internal/app/groupcalls/service.go @@ -6,9 +6,12 @@ package groupcalls import ( "context" "crypto/rand" + "crypto/subtle" "encoding/base64" "encoding/binary" "fmt" + "strconv" + "strings" "telesrv/internal/domain" "telesrv/internal/store" @@ -24,8 +27,10 @@ func NewService(st store.GroupCallStore) *Service { return &Service{store: st} } -// Create 分配 id/access_hash 并建会。 -func (s *Service) Create(ctx context.Context, channelID, creatorUserID int64, title string, now int) (domain.GroupCall, error) { +// Create 分配 id/access_hash 并建会。rtmpStream=true 创建 RTMP 直播房间； +// joinMuted=true（广播频道直播）让非管理员入会即被静音且不可自解； +// scheduleDate>0 创建定时通话（客户端倒计时等待 startScheduled）。 +func (s *Service) Create(ctx context.Context, channelID, creatorUserID int64, title string, rtmpStream, joinMuted bool, scheduleDate, now int) (domain.GroupCall, error) { id, err := randomPositiveInt64() if err != nil { return domain.GroupCall{}, err @@ -40,11 +45,73 @@ func (s *Service) Create(ctx context.Context, channelID, creatorUserID int64, ti ChannelID: channelID, CreatorUserID: creatorUserID, Title: title, + RtmpStream: rtmpStream, + JoinMuted: joinMuted, + ScheduleDate: scheduleDate, Version: 1, CreatedAt: now, }) } +// StartScheduled 把定时通话转为进行中（清 schedule_date）；changed=false 幂等。 +func (s *Service) StartScheduled(ctx context.Context, callID int64) (domain.GroupCall, bool, error) { + return s.store.StartScheduledGroupCall(ctx, callID) +} + +// SetScheduleSubscription 写入/清除开播提醒订阅。 +func (s *Service) SetScheduleSubscription(ctx context.Context, callID, userID int64, subscribed bool) error { + return s.store.SetScheduleStartSubscription(ctx, callID, userID, subscribed) +} + +// ScheduleSubscriberIDs 返回订阅开播提醒的 userID。 +func (s *Service) ScheduleSubscriberIDs(ctx context.Context, callID int64) ([]int64, error) { + return s.store.ListScheduleSubscriberIDs(ctx, callID) +} + +// RtmpStreamKey 返回 channel 的持久 RTMP 推流密钥；不存在或 rotate=true 时生成 +// 新 key（覆盖写入，旧 key 即刻失效）。key 形如 "_"，ingest 端 +// 据前缀定位 channel、再整串比对鉴权。 +func (s *Service) RtmpStreamKey(ctx context.Context, channelID int64, rotate bool, now int) (string, error) { + if !rotate { + key, found, err := s.store.GetRtmpStreamKey(ctx, channelID) + if err != nil { + return "", err + } + if found { + return key, nil + } + } + var buf [24]byte + if _, err := rand.Read(buf[:]); err != nil { + return "", fmt.Errorf("groupcalls: random rtmp key: %w", err) + } + key := fmt.Sprintf("%d_%x", channelID, buf) + if err := s.store.SetRtmpStreamKey(ctx, channelID, key, now); err != nil { + return "", err + } + return key, nil +} + +// VerifyRtmpStreamKey 校验推流密钥并返回其所属 channelID（RTMP ingest 鉴权入口）。 +func (s *Service) VerifyRtmpStreamKey(ctx context.Context, key string) (int64, bool, error) { + sep := strings.IndexByte(key, '_') + if sep <= 0 { + return 0, false, nil + } + channelID, err := strconv.ParseInt(key[:sep], 10, 64) + if err != nil || channelID <= 0 { + return 0, false, nil + } + stored, found, err := s.store.GetRtmpStreamKey(ctx, channelID) + if err != nil { + return 0, false, err + } + if !found || subtle.ConstantTimeCompare([]byte(stored), []byte(key)) != 1 { + return 0, false, nil + } + return channelID, true, nil +} + // CreateConference 分配 id/access_hash/slug 并创建 ad-hoc conference call。 func (s *Service) CreateConference(ctx context.Context, creatorUserID, randomID, migratedFromPhoneCallID int64, now int) (domain.GroupCall, error) { for i := 0; i < 8; i++ { diff --git a/internal/app/livestream/part.go b/internal/app/livestream/part.go new file mode 100644 index 00000000..c85d47fe --- /dev/null +++ b/internal/app/livestream/part.go @@ -0,0 +1,70 @@ +package livestream + +import "encoding/binary" + +// tgcalls broadcast part 打包（消费方 tgcalls VideoStreamingPart.cpp +// consumeVideoStreamInfo）。unified（RTMP）模式的 part 结构： +// +// int32(LE) 签名 0xa12e810d +// TL 风格短字符串：容器名（本实现恒 "mp4"） +// int32 activeMask（单一 unified 轨恒 1） +// int32 eventCount（消费方只读第一个 event，恒写 1） +// event: int32 offset(=0，相对头部之后的数据) + 字符串 endpointId("unified") +// + int32 rotation(0) + int32 extra(0) +// 随后紧跟容器数据（音视频同容器，客户端分别按 Video/Audio content type 解）。 +const partSignature uint32 = 0xa12e810d + +// partContainer 必须落在 TDesktop 裁剪版 ffmpeg 的 demuxer 白名单内 +// （Telegram/build/prepare/prepare.py --enable-demuxer=...，含 mov/mp4、无 mpegts）。 +// "mp4" 由 mov demuxer 别名匹配；tgcalls AVIO 支持 seek，完整 mp4（moov 在尾）可解。 +const ( + partContainer = "mp4" + partEndpointID = "unified" +) + +// appendTLString 按 tgcalls readSerializedString 的逆操作写入字符串： +// 长度 <254 用 1 字节长度 + 数据，整体（含长度字节）补齐到 4 字节； +// 否则 0xFE + 3 字节小端长度 + 数据，数据补齐到 4 字节。 +func appendTLString(dst []byte, s string) []byte { + n := len(s) + if n < 254 { + dst = append(dst, byte(n)) + dst = append(dst, s...) + for (n+1)%4 != 0 { + dst = append(dst, 0) + n++ + } + return dst + } + dst = append(dst, 0xFE, byte(n), byte(n>>8), byte(n>>16)) + dst = append(dst, s...) + for n%4 != 0 { + dst = append(dst, 0) + n++ + } + return dst +} + +func appendUint32(dst []byte, v uint32) []byte { + var buf [4]byte + binary.LittleEndian.PutUint32(buf[:], v) + return append(dst, buf[:]...) +} + +func appendInt32(dst []byte, v int32) []byte { + return appendUint32(dst, uint32(v)) +} + +// packUnifiedPart 把一段自包含 MPEG-TS 数据包成 tgcalls broadcast part。 +func packUnifiedPart(tsData []byte) []byte { + out := make([]byte, 0, len(tsData)+48) + out = appendUint32(out, partSignature) + out = appendTLString(out, partContainer) + out = appendInt32(out, 1) // activeMask + out = appendInt32(out, 1) // eventCount + out = appendInt32(out, 0) // event.offset + out = appendTLString(out, partEndpointID) + out = appendInt32(out, 0) // event.rotation + out = appendInt32(out, 0) // event.extra + return append(out, tsData...) +} diff --git a/internal/app/livestream/part_test.go b/internal/app/livestream/part_test.go new file mode 100644 index 00000000..f9cbf902 --- /dev/null +++ b/internal/app/livestream/part_test.go @@ -0,0 +1,103 @@ +package livestream + +import ( + "encoding/binary" + "testing" +) + +// readTLString 复刻 tgcalls readSerializedString，用于反解 packUnifiedPart 的头。 +func readTLString(data []byte, off *int) (string, bool) { + if *off >= len(data) { + return "", false + } + first := int(data[*off]) + *off++ + var length, padding int + if first == 254 { + if *off+3 > len(data) { + return "", false + } + length = int(data[*off]) | int(data[*off+1])<<8 | int(data[*off+2])<<16 + *off += 3 + padding = (4 - length%4) % 4 + } else { + length = first + padding = (4 - (length+1)%4) % 4 + } + if *off+length > len(data) { + return "", false + } + s := string(data[*off : *off+length]) + *off += length + padding + return s, true +} + +func readI32(data []byte, off *int) (int32, bool) { + if *off+4 > len(data) { + return 0, false + } + v := int32(binary.LittleEndian.Uint32(data[*off : *off+4])) + *off += 4 + return v, true +} + +// TestPackUnifiedPartMatchesTgcallsHeader 校验打包头逐字段可被 tgcalls +// consumeVideoStreamInfo 解出：签名、容器名、activeMask、单 event(endpoint="unified")， +// 且 event.offset=0 对应紧随其后的 TS 数据。 +func TestPackUnifiedPartMatchesTgcallsHeader(t *testing.T) { + ts := []byte{0x47, 0x40, 0x00, 0x10, 0xDE, 0xAD, 0xBE, 0xEF} // 伪 TS 数据 + part := packUnifiedPart(ts) + + off := 0 + sig, ok := readI32(part, &off) + if !ok || uint32(sig) != partSignature { + t.Fatalf("signature = %#x ok=%v, want %#x", uint32(sig), ok, partSignature) + } + container, ok := readTLString(part, &off) + if !ok || container != partContainer { + t.Fatalf("container = %q ok=%v, want %q", container, ok, partContainer) + } + activeMask, ok := readI32(part, &off) + if !ok || activeMask != 1 { + t.Fatalf("activeMask = %d ok=%v, want 1", activeMask, ok) + } + eventCount, ok := readI32(part, &off) + if !ok || eventCount != 1 { + t.Fatalf("eventCount = %d ok=%v, want 1", eventCount, ok) + } + eventOffset, ok := readI32(part, &off) + if !ok || eventOffset != 0 { + t.Fatalf("event.offset = %d ok=%v, want 0", eventOffset, ok) + } + endpoint, ok := readTLString(part, &off) + if !ok || endpoint != partEndpointID { + t.Fatalf("endpoint = %q ok=%v, want %q", endpoint, ok, partEndpointID) + } + rotation, _ := readI32(part, &off) + extra, _ := readI32(part, &off) + if rotation != 0 || extra != 0 { + t.Fatalf("rotation/extra = %d/%d, want 0/0", rotation, extra) + } + // 头之后（event.offset=0 起）应为原始 TS 数据。 + if got := part[off:]; string(got) != string(ts) { + t.Fatalf("payload = %x, want %x", got, ts) + } +} + +// TestAppendTLStringPadding 校验短/长字符串都补齐到 4 字节边界。 +func TestAppendTLStringPadding(t *testing.T) { + for _, s := range []string{"", "a", "ab", "abc", "mpegts", "unified"} { + out := appendTLString(nil, s) + if len(out)%4 != 0 { + t.Fatalf("appendTLString(%q) len=%d not 4-aligned", s, len(out)) + } + off := 0 + got, ok := readTLString(out, &off) + if !ok || got != s { + t.Fatalf("roundtrip %q -> %q ok=%v", s, got, ok) + } + if off != len(out) { + t.Fatalf("roundtrip %q consumed %d of %d bytes", s, off, len(out)) + } + } +} diff --git a/internal/app/livestream/rtmp.go b/internal/app/livestream/rtmp.go new file mode 100644 index 00000000..e1b52fb9 --- /dev/null +++ b/internal/app/livestream/rtmp.go @@ -0,0 +1,68 @@ +package livestream + +import ( + "context" + "io" + "net" + + rtmp "github.com/yutopp/go-rtmp" + rtmpmsg "github.com/yutopp/go-rtmp/message" + "go.uber.org/zap" +) + +// rtmpHandler 是单条 RTMP 连接的回调：publish 时用 stream key 鉴权并绑定 +// channel，media tag 直通 FLV → ffmpeg。一条连接只允许一路 publish。 +type rtmpHandler struct { + rtmp.DefaultHandler + svc *Service + conn net.Conn + stream *stream +} + +func (h *rtmpHandler) OnPublish(_ *rtmp.StreamContext, _ uint32, cmd *rtmpmsg.NetStreamPublish) error { + if h.stream != nil { + return errPublishRejected + } + st, err := h.svc.startPublish(context.Background(), cmd.PublishingName, h.conn) + if err != nil { + h.svc.log.Warn("rtmp publish rejected", zap.Error(err)) + return errPublishRejected + } + h.stream = st + return nil +} + +func (h *rtmpHandler) OnSetDataFrame(timestamp uint32, data *rtmpmsg.NetStreamSetDataFrame) error { + if h.stream == nil { + return nil + } + // onMetaData 原样透传给 ffmpeg（可选信息，写失败不断流）。 + _ = h.stream.writeTag(18, timestamp, data.Payload) + return nil +} + +func (h *rtmpHandler) OnAudio(timestamp uint32, payload io.Reader) error { + return h.writeMedia(8, timestamp, payload) +} + +func (h *rtmpHandler) OnVideo(timestamp uint32, payload io.Reader) error { + return h.writeMedia(9, timestamp, payload) +} + +func (h *rtmpHandler) writeMedia(tagType byte, timestamp uint32, payload io.Reader) error { + if h.stream == nil { + return nil + } + body, err := io.ReadAll(payload) + if err != nil { + return err + } + return h.stream.writeTag(tagType, timestamp, body) +} + +func (h *rtmpHandler) OnClose() { + if h.stream != nil { + h.svc.endPublish(h.stream) + h.stream = nil + } +} diff --git a/internal/app/livestream/segmenter.go b/internal/app/livestream/segmenter.go new file mode 100644 index 00000000..263c8ccd --- /dev/null +++ b/internal/app/livestream/segmenter.go @@ -0,0 +1,285 @@ +package livestream + +import ( + "bufio" + "encoding/binary" + "fmt" + "io" + "os" + "os/exec" + "path/filepath" + "sync" + + "go.uber.org/zap" + + "telesrv/internal/domain" +) + +// segmentDurationMs 是 broadcast part 的固定时长。tgcalls StreamingMediaContext +// 写死 _segmentDuration=1000（scale 0），时间轴推进按 +1000 走，segment 必须严格 +// 1 秒切齐（转码强制每秒关键帧保证切点）。 +const segmentDurationMs = 1000 + +// minSegmentsBeforeAnnounce：客户端拿到 last_timestamp_ms 后从 last-2000 开始拉， +// 至少积 3 段再对外公布时间轴，避免起播即请求不存在的负偏移段。 +const minSegmentsBeforeAnnounce = 3 + +// stream 是一路活跃 RTMP 推流：FLV 入 ffmpeg（转码+按秒切 MPEG-TS）→ 打包 part +// 入内存 ring。时间轴：T0 取首段完成时刻向下取整秒，第 i 段的 time_ms = T0+i*1000。 +type stream struct { + channelID int64 + log *zap.Logger + dir string + keep int + + cmd *exec.Cmd + stdin io.WriteCloser + closer io.Closer // RTMP 连接，DropChannel 时踢掉推流端 + + mu sync.Mutex + flvStarted bool + baseMs int64 // T0；0=尚未产出任何 segment + segments map[int64][]byte // time_ms → packed part + order []int64 // 按 time_ms 升序（ring 淘汰用） + lastMs int64 // 最新 segment 的 time_ms + segmentSeq int64 // 已完成 segment 计数 + ended bool + oversizeWas bool + + nowMs func() int64 +} + +// ffmpegArgs 组装转码+切段命令。要点： +// - 强制每秒关键帧（-force_key_frames）保证 -f segment 严格按 1s 切； +// - 严格码率上限：TDesktop 拉 part 单次 `upload.getFile(offset=0,limit=128KiB)` +// 且**不续读**，单段（视频+音频+TS 开销）>128KiB 会被静默截断致花屏。 +// 故 unified 单质量必须压在 ~1Mbps 以下——这里目标 ~640kbps：视频 +// 480k（maxrate/bufsize=480k 收紧到每秒 VBV，杜绝关键帧段爆量）+ 音频 64k， +// 并降到 640x360/24fps 进一步留余量； +// - 输出自包含 mp4（每段独立 moov，可单独 avformat_open_input）。⚠ 不能用 +// MPEG-TS：TDesktop 裁剪版 ffmpeg 的 demuxer 白名单只有 mov/mp4 系 +// （prepare.py --enable-demuxer），mpegts 会让 tgcalls 打不开容器 → 黑屏； +// - -segment_list pipe:1 每完成一段输出一行文件名，作为完成事件。 +func ffmpegArgs(outDir string) []string { + return []string{ + "-hide_banner", "-nostats", "-loglevel", "warning", + "-fflags", "+genpts", + "-f", "flv", "-i", "pipe:0", + "-vf", "scale=-2:360", "-r", "24", + "-c:v", "libx264", "-preset", "veryfast", "-tune", "zerolatency", + "-profile:v", "main", "-pix_fmt", "yuv420p", + "-b:v", "480k", "-maxrate", "480k", "-bufsize", "480k", + "-g", "24", "-keyint_min", "24", + "-force_key_frames", "expr:gte(t,n_forced*1)", "-sc_threshold", "0", + "-c:a", "aac", "-b:a", "64k", "-ar", "48000", "-ac", "2", + "-f", "segment", + "-segment_time", "1", + "-segment_format", "mp4", + "-segment_format_options", "movflags=+faststart", + "-segment_list", "pipe:1", + "-segment_list_type", "flat", + "-reset_timestamps", "1", + filepath.Join(outDir, "seg%06d.mp4"), + } +} + +func newStream(channelID int64, ffmpegPath, workDir string, keep int, closer io.Closer, nowMs func() int64, log *zap.Logger) (*stream, error) { + dir, err := os.MkdirTemp(workDir, fmt.Sprintf("live_%d_", channelID)) + if err != nil { + return nil, fmt.Errorf("livestream: workdir: %w", err) + } + cmd := exec.Command(ffmpegPath, ffmpegArgs(dir)...) + stdin, err := cmd.StdinPipe() + if err != nil { + return nil, fmt.Errorf("livestream: ffmpeg stdin: %w", err) + } + stdout, err := cmd.StdoutPipe() + if err != nil { + return nil, fmt.Errorf("livestream: ffmpeg stdout: %w", err) + } + stderr, err := cmd.StderrPipe() + if err != nil { + return nil, fmt.Errorf("livestream: ffmpeg stderr: %w", err) + } + if err := cmd.Start(); err != nil { + return nil, fmt.Errorf("livestream: start ffmpeg: %w", err) + } + s := &stream{ + channelID: channelID, + log: log, + dir: dir, + keep: keep, + cmd: cmd, + stdin: stdin, + closer: closer, + segments: make(map[int64][]byte), + nowMs: nowMs, + } + go s.readSegmentList(stdout) + go s.logStderr(stderr) + go func() { + _ = cmd.Wait() + s.mu.Lock() + s.ended = true + s.mu.Unlock() + }() + return s, nil +} + +// readSegmentList 消费 ffmpeg 的 segment 完成事件流。 +func (s *stream) readSegmentList(r io.Reader) { + scanner := bufio.NewScanner(r) + for scanner.Scan() { + name := scanner.Text() + if name == "" { + continue + } + path := filepath.Join(s.dir, filepath.Base(name)) + data, err := os.ReadFile(path) + if err != nil { + s.log.Warn("live stream read segment", zap.String("path", path), zap.Error(err)) + continue + } + // 诊断：TELESRV_LIVESTREAM_DUMP_DIR 非空时把原始 TS 切片留档供 ffprobe 检查。 + if dump := os.Getenv("TELESRV_LIVESTREAM_DUMP_DIR"); dump != "" { + _ = os.MkdirAll(dump, 0o755) + _ = os.WriteFile(filepath.Join(dump, fmt.Sprintf("ch%d_%s", s.channelID, filepath.Base(name))), data, 0o644) + } + _ = os.Remove(path) + s.addSegment(data) + } +} + +func (s *stream) logStderr(r io.Reader) { + scanner := bufio.NewScanner(r) + for scanner.Scan() { + s.log.Info("ffmpeg", zap.Int64("channel_id", s.channelID), zap.String("line", scanner.Text())) + } +} + +func (s *stream) addSegment(tsData []byte) { + part := packUnifiedPart(tsData) + s.mu.Lock() + defer s.mu.Unlock() + if s.baseMs == 0 { + s.baseMs = s.nowMs() / segmentDurationMs * segmentDurationMs + } + timeMs := s.baseMs + s.segmentSeq*segmentDurationMs + s.segmentSeq++ + s.segments[timeMs] = part + s.order = append(s.order, timeMs) + s.lastMs = timeMs + for len(s.order) > s.keep { + delete(s.segments, s.order[0]) + s.order = s.order[1:] + } + s.log.Debug("live stream segment produced", + zap.Int64("channel_id", s.channelID), zap.Int64("time_ms", timeMs), + zap.Int("bytes", len(part)), zap.Int64("seq", s.segmentSeq), + zap.Int64("wall_ms", s.nowMs())) + if len(part) > 128<<10 && !s.oversizeWas { + s.oversizeWas = true + s.log.Warn("live stream segment exceeds 128KiB, client will truncate", + zap.Int64("channel_id", s.channelID), zap.Int("bytes", len(part))) + } +} + +// channels 返回当前时间轴（不足 minSegmentsBeforeAnnounce 段时不公布）。 +func (s *stream) channels() []domain.LiveStreamChannel { + s.mu.Lock() + defer s.mu.Unlock() + if s.ended || len(s.order) < minSegmentsBeforeAnnounce { + return nil + } + return []domain.LiveStreamChannel{{Channel: 1, Scale: 0, LastTimestampMs: s.lastMs}} +} + +// part 取指定 time_ms 的打包 part。 +func (s *stream) part(timeMs int64) ([]byte, error) { + s.mu.Lock() + defer s.mu.Unlock() + if s.baseMs == 0 { + return nil, domain.ErrLiveStreamPartNotReady + } + if timeMs < s.baseMs || (timeMs-s.baseMs)%segmentDurationMs != 0 { + return nil, domain.ErrLiveStreamPartExpired + } + if timeMs > s.lastMs { + if s.ended { + return nil, domain.ErrLiveStreamNoStream + } + return nil, domain.ErrLiveStreamPartNotReady + } + part, ok := s.segments[timeMs] + if !ok { + return nil, domain.ErrLiveStreamPartExpired + } + return part, nil +} + +func (s *stream) active() bool { + s.mu.Lock() + defer s.mu.Unlock() + return !s.ended +} + +// stop 结束推流：断 RTMP 连接、关 ffmpeg stdin（自然退出），清空缓冲目录。 +func (s *stream) stop() { + s.mu.Lock() + if s.ended { + s.mu.Unlock() + return + } + s.ended = true + s.mu.Unlock() + if s.closer != nil { + _ = s.closer.Close() + } + _ = s.stdin.Close() + go func() { + _ = s.cmd.Wait() + _ = os.RemoveAll(s.dir) + }() +} + +// ---- FLV 写入（RTMP tag → ffmpeg stdin）---- + +var flvHeader = []byte{'F', 'L', 'V', 0x01, 0x05, 0x00, 0x00, 0x00, 0x09, 0x00, 0x00, 0x00, 0x00} + +// writeTag 把一条 RTMP media/data tag 以 FLV 封装写进 ffmpeg stdin。 +// tagType：8=audio 9=video 18=script data。 +func (s *stream) writeTag(tagType byte, timestampMs uint32, body []byte) error { + s.mu.Lock() + if s.ended { + s.mu.Unlock() + return domain.ErrLiveStreamNoStream + } + started := s.flvStarted + s.flvStarted = true + s.mu.Unlock() + if !started { + if _, err := s.stdin.Write(flvHeader); err != nil { + return err + } + } + var hdr [11]byte + hdr[0] = tagType + hdr[1] = byte(len(body) >> 16) + hdr[2] = byte(len(body) >> 8) + hdr[3] = byte(len(body)) + hdr[4] = byte(timestampMs >> 16) + hdr[5] = byte(timestampMs >> 8) + hdr[6] = byte(timestampMs) + hdr[7] = byte(timestampMs >> 24) + // stream id hdr[8:11] = 0 + if _, err := s.stdin.Write(hdr[:]); err != nil { + return err + } + if _, err := s.stdin.Write(body); err != nil { + return err + } + var prev [4]byte + binary.BigEndian.PutUint32(prev[:], uint32(11+len(body))) + _, err := s.stdin.Write(prev[:]) + return err +} diff --git a/internal/app/livestream/service.go b/internal/app/livestream/service.go new file mode 100644 index 00000000..bdc501d4 --- /dev/null +++ b/internal/app/livestream/service.go @@ -0,0 +1,188 @@ +// Package livestream 实现频道 RTMP 直播的媒体面：RTMP ingest（OBS 推流）→ +// ffmpeg 转码按秒切段 → tgcalls broadcast part 内存 ring → 观众经 +// upload.getFile(inputGroupCallStream) 拉流。信令面（groupCall/participants） +// 仍归 app/groupcalls；本包只认 stream key ↔ channelID 绑定。 +// +// 定位：dev 主路径（单实例、内存 ring、无 CDN/多码率）， +// 生产级转码集群与分发留后续任务（见 docs/voip-module.md 直播小节）。 +package livestream + +import ( + "context" + "errors" + "fmt" + "io" + "net" + "os" + "sync" + "time" + + rtmp "github.com/yutopp/go-rtmp" + "go.uber.org/zap" + + "telesrv/internal/domain" +) + +var errPublishRejected = errors.New("livestream: publish rejected") + +// KeyResolver 校验 RTMP 推流密钥并返回其绑定的 channelID（app/groupcalls 实现）。 +type KeyResolver interface { + VerifyRtmpStreamKey(ctx context.Context, key string) (channelID int64, ok bool, err error) +} + +// Config 是直播媒体面配置。 +type Config struct { + // ListenAddr 是 RTMP ingest 监听地址（如 ":2400"）。 + ListenAddr string + // FFmpegPath 是 ffmpeg 可执行文件路径（默认 "ffmpeg"，走 PATH）。 + FFmpegPath string + // WorkDir 是切段临时目录（默认系统临时目录）。 + WorkDir string + // SegmentKeep 是每路流内存保留的 segment 数（秒），默认 32。 + SegmentKeep int +} + +// Service 管理全部活跃推流会话，并向 rpc 层提供拉流查询。 +type Service struct { + cfg Config + keys KeyResolver + log *zap.Logger + + mu sync.Mutex + streams map[int64]*stream // channelID → 活跃流 + listener net.Listener +} + +// NewService 创建直播服务（不监听；Start 启动 ingest）。 +func NewService(cfg Config, keys KeyResolver, log *zap.Logger) *Service { + if cfg.FFmpegPath == "" { + cfg.FFmpegPath = "ffmpeg" + } + if cfg.SegmentKeep <= 0 { + cfg.SegmentKeep = 32 + } + if cfg.WorkDir == "" { + cfg.WorkDir = os.TempDir() + } + return &Service{cfg: cfg, keys: keys, log: log, streams: make(map[int64]*stream)} +} + +// Start 启动 RTMP ingest 监听。 +func (s *Service) Start() error { + ln, err := net.Listen("tcp", s.cfg.ListenAddr) + if err != nil { + return fmt.Errorf("livestream: listen rtmp %s: %w", s.cfg.ListenAddr, err) + } + s.mu.Lock() + s.listener = ln + s.mu.Unlock() + srv := rtmp.NewServer(&rtmp.ServerConfig{ + OnConnect: func(conn net.Conn) (io.ReadWriteCloser, *rtmp.ConnConfig) { + return conn, &rtmp.ConnConfig{ + Handler: &rtmpHandler{svc: s, conn: conn}, + ControlState: rtmp.StreamControlStateConfig{ + DefaultBandwidthWindowSize: 6 * 1024 * 1024 / 8, + }, + } + }, + }) + go func() { + if err := srv.Serve(ln); err != nil { + s.log.Warn("rtmp server exited", zap.Error(err)) + } + }() + s.log.Info("live stream rtmp ingest listening", zap.String("addr", s.cfg.ListenAddr)) + return nil +} + +// Close 停止监听并结束全部推流。 +func (s *Service) Close() { + s.mu.Lock() + ln := s.listener + streams := make([]*stream, 0, len(s.streams)) + for _, st := range s.streams { + streams = append(streams, st) + } + s.streams = make(map[int64]*stream) + s.mu.Unlock() + if ln != nil { + _ = ln.Close() + } + for _, st := range streams { + st.stop() + } +} + +// startPublish 鉴权 stream key 并建立一路新流；同 channel 已有活跃流时顶掉旧流 +// （OBS 断线重连的自然语义）。 +func (s *Service) startPublish(ctx context.Context, key string, conn net.Conn) (*stream, error) { + channelID, ok, err := s.keys.VerifyRtmpStreamKey(ctx, key) + if err != nil { + return nil, err + } + if !ok { + return nil, fmt.Errorf("livestream: bad stream key") + } + st, err := newStream(channelID, s.cfg.FFmpegPath, s.cfg.WorkDir, s.cfg.SegmentKeep, conn, + func() int64 { return time.Now().UnixMilli() }, s.log) + if err != nil { + return nil, err + } + s.mu.Lock() + old := s.streams[channelID] + s.streams[channelID] = st + s.mu.Unlock() + if old != nil { + old.stop() + } + s.log.Info("live stream publish started", zap.Int64("channel_id", channelID), + zap.String("remote", conn.RemoteAddr().String())) + return st, nil +} + +// endPublish 在推流连接断开时收尾（仅当它仍是当前流时移除）。 +func (s *Service) endPublish(st *stream) { + s.mu.Lock() + if s.streams[st.channelID] == st { + delete(s.streams, st.channelID) + } + s.mu.Unlock() + st.stop() + s.log.Info("live stream publish ended", zap.Int64("channel_id", st.channelID)) +} + +// StreamChannels 返回 channel 当前直播时间轴；无活跃推流返回空。 +func (s *Service) StreamChannels(channelID int64) []domain.LiveStreamChannel { + s.mu.Lock() + st := s.streams[channelID] + s.mu.Unlock() + if st == nil || !st.active() { + return nil + } + return st.channels() +} + +// StreamPart 按 time_ms 取打包好的 broadcast part（仅 scale 0）。 +func (s *Service) StreamPart(channelID int64, timeMs int64, scale int) ([]byte, error) { + if scale != 0 { + return nil, domain.ErrLiveStreamPartExpired + } + s.mu.Lock() + st := s.streams[channelID] + s.mu.Unlock() + if st == nil { + return nil, domain.ErrLiveStreamNoStream + } + return st.part(timeMs) +} + +// DropChannel 断开该 channel 的推流会话并清空缓冲（discard 直播 / revoke key）。 +func (s *Service) DropChannel(channelID int64) { + s.mu.Lock() + st := s.streams[channelID] + delete(s.streams, channelID) + s.mu.Unlock() + if st != nil { + st.stop() + } +} diff --git a/internal/config/config.go b/internal/config/config.go index 9022b26a..ca33f341 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -234,6 +234,19 @@ type Config struct { // CallForceRelay 强制 p2p_allowed=false（调试 TURN 中继路径用）。 CallForceRelay bool + // LiveStreamEnable 为 true 时启用频道 RTMP 直播媒体面（内嵌 RTMP ingest + ffmpeg 切段）。 + LiveStreamEnable bool + // LiveStreamRtmpAddr 是 RTMP ingest 的 TCP 监听地址（默认 ":2400"）。 + LiveStreamRtmpAddr string + // LiveStreamRtmpURL 是返回给推流端（OBS）的服务器地址；为空回落 rtmp://:2400/live。 + LiveStreamRtmpURL string + // LiveStreamFFmpegPath 是 ffmpeg 可执行路径（默认走 PATH 的 "ffmpeg"）。 + LiveStreamFFmpegPath string + // LiveStreamWorkDir 是切段临时目录（默认系统临时目录）。 + LiveStreamWorkDir string + // LiveStreamSegmentKeep 是每路流内存保留的 segment 秒数（默认 32）。 + LiveStreamSegmentKeep int + // SFUEnable 为 false 时群通话只走信令（M0 模式，无媒体）。 SFUEnable bool // SFUUDPPort 是内嵌 SFU 的单 UDP 端口（pion ICE UDPMux）。Windows 防火墙需放行。 @@ -380,6 +393,13 @@ func Load() (Config, error) { SFUEnable: envBoolOr("TELESRV_SFU_ENABLE", true), SFUUDPPort: envIntOr("TELESRV_SFU_UDP_PORT", 12399), SFUAdvertiseIP: envOr("TELESRV_SFU_ADVERTISE_IP", ""), + + LiveStreamEnable: envBoolOr("TELESRV_LIVESTREAM_ENABLE", true), + LiveStreamRtmpAddr: envOr("TELESRV_LIVESTREAM_RTMP_ADDR", ":2400"), + LiveStreamRtmpURL: envOr("TELESRV_LIVESTREAM_RTMP_URL", ""), + LiveStreamFFmpegPath: envOr("TELESRV_LIVESTREAM_FFMPEG_PATH", "ffmpeg"), + LiveStreamWorkDir: envOr("TELESRV_LIVESTREAM_WORK_DIR", ""), + LiveStreamSegmentKeep: envIntOr("TELESRV_LIVESTREAM_SEGMENT_KEEP", 32), } return cfg, nil } diff --git a/internal/domain/channel.go b/internal/domain/channel.go index 5ca609fe..418fe7b9 100644 --- a/internal/domain/channel.go +++ b/internal/domain/channel.go @@ -530,6 +530,9 @@ const ( // ChannelActionGroupCall 映射 messageActionGroupCall：started（CallDuration=0） // 与 ended（CallDuration>0）共用同一构造器，官方语义即如此。 ChannelActionGroupCall ChannelMessageActionType = "group_call" + // ChannelActionGroupCallScheduled 映射 messageActionGroupCallScheduled： + // 定时通话创建的服务消息（"scheduled a video chat for ..."），ScheduleDate 必填。 + ChannelActionGroupCallScheduled ChannelMessageActionType = "group_call_scheduled" // ChannelActionInviteToGroupCall 映射 messageActionInviteToGroupCall（被邀请 // 者通过频道消息收到可点击的入会卡片，UserIDs 为受邀人）。 ChannelActionInviteToGroupCall ChannelMessageActionType = "invite_to_group_call" @@ -561,6 +564,8 @@ type ChannelMessageAction struct { CallID int64 CallAccessHash int64 CallDuration int + // CallScheduleDate 仅 group_call_scheduled 使用（开播时间）。 + CallScheduleDate int // Boosts 仅 boost_apply 服务消息使用。 Boosts int // BroadcastMessagesAllowed/Stars 仅 paid_messages_price 服务消息使用。 diff --git a/internal/domain/groupcall.go b/internal/domain/groupcall.go index 1ed06216..4c3044e9 100644 --- a/internal/domain/groupcall.go +++ b/internal/domain/groupcall.go @@ -55,6 +55,15 @@ type GroupCall struct { State GroupCallState Title string JoinMuted bool + // RtmpStream=true 表示这是 RTMP 直播房间（推流走外部 RTMP ingest，观众经 + // broadcast 拉流），join 不建 SFU 连接、connection params 返回 stream JSON。 + RtmpStream bool + // ScheduleDate>0 表示定时通话尚未开始（客户端显示倒计时面板、不入会）； + // startScheduledGroupCall 清零后客户端自动 initialJoin。持久列。 + ScheduleDate int + // ScheduleStartSubscribed 是 per-viewer 投影字段（非持久列）：当前 viewer 是否 + // 订阅了开播提醒。RPC/推送出站前按 viewer 回填（与 Creator flag 同类语义）。 + ScheduleStartSubscribed bool // Version 是参与者协议版本：所有参与者变更事务内 +1，单调且持久。 Version int ParticipantsCount int @@ -87,6 +96,10 @@ func (c GroupCall) Conference() bool { type GroupCallParticipant struct { CallID int64 UserID int64 + // JoinAsChannelID 非零表示以该频道/群身份入会（匿名管理员语义），TL 输出 + // participant.peer=PeerChannel；0=以本人用户身份。身份唯一键仍是 UserID， + // 换身份 rejoin 是替换。 + JoinAsChannelID int64 // SSRC 是客户端在 join JSON 里自报的 audio ssrc（uint32 值域，存 int64 防符号坑）。 SSRC int64 JoinDate int @@ -124,13 +137,15 @@ type CreateGroupCallRequest struct { // JoinGroupCallRequest 加入/重进群通话（rejoin 同主键换新 ssrc）。 type JoinGroupCallRequest struct { - CallID int64 - UserID int64 - SSRC int64 - Muted bool - IsAdmin bool - PublicKey []byte - JoinBlock []byte + CallID int64 + UserID int64 + // JoinAsChannelID 非零=以频道身份入会（调用方已完成权限校验）。 + JoinAsChannelID int64 + SSRC int64 + Muted bool + IsAdmin bool + PublicKey []byte + JoinBlock []byte // VideoJSON 是本次 join 铸造的视频内部状态（endpoint+源组+active）；rejoin // 整体替换并**清空旧 PresentationJSON**（客户端主连接 rejoin 后会重发 // joinGroupCallPresentation，旧屏幕登记必须作废）。 diff --git a/internal/domain/livestream.go b/internal/domain/livestream.go new file mode 100644 index 00000000..f2060f67 --- /dev/null +++ b/internal/domain/livestream.go @@ -0,0 +1,20 @@ +package domain + +import "errors" + +// LiveStreamChannel 是直播时间轴上一个可拉流通道的快照（RTMP unified 模式恒为 +// channel=1 / scale=0；last_timestamp_ms 是最新可取 segment 的起始时间戳）。 +type LiveStreamChannel struct { + Channel int + Scale int + LastTimestampMs int64 +} + +// 直播拉流业务错误；rpc 层映射见 upload.getFile(inputGroupCallStream)： +// - PartNotReady → TIME_TOO_BIG（客户端 100ms 后原样重试） +// - PartExpired / NoStream → 普通 400（客户端重新对时 resync） +var ( + ErrLiveStreamPartNotReady = errors.New("live stream part not ready") + ErrLiveStreamPartExpired = errors.New("live stream part expired") + ErrLiveStreamNoStream = errors.New("live stream not active") +) diff --git a/internal/rpc/convert_channels_core.go b/internal/rpc/convert_channels_core.go index abe9af95..dab03787 100644 --- a/internal/rpc/convert_channels_core.go +++ b/internal/rpc/convert_channels_core.go @@ -258,6 +258,11 @@ func tgChannelMessageAction(action domain.ChannelMessageAction) tg.MessageAction out.SetDuration(action.CallDuration) } return out + case domain.ChannelActionGroupCallScheduled: + return &tg.MessageActionGroupCallScheduled{ + Call: &tg.InputGroupCall{ID: action.CallID, AccessHash: action.CallAccessHash}, + ScheduleDate: action.CallScheduleDate, + } case domain.ChannelActionInviteToGroupCall: return &tg.MessageActionInviteToGroupCall{ Call: &tg.InputGroupCall{ID: action.CallID, AccessHash: action.CallAccessHash}, diff --git a/internal/rpc/convert_phone_group.go b/internal/rpc/convert_phone_group.go index 33f347d1..a7b7e448 100644 --- a/internal/rpc/convert_phone_group.go +++ b/internal/rpc/convert_phone_group.go @@ -20,6 +20,11 @@ import ( // this chat」即此（TDesktop emitShareScreenError / DrKLO ChatObject.canStreamVideo）。 const groupCallUnmutedVideoLimit = 30 +// groupCallStreamDCID 是写进 RTMP groupCall.stream_dc_id 的 DC 号。本服务单 DC， +// 由 router New() 从 cfg.DC 初始化；缺省 2 与 help.getConfig 的 ThisDC 默认一致。 +// 值仅用于客户端拉流 DC 路由标记（telesrv 拉流走同连接），避免 TDesktop fallback 日志。 +var groupCallStreamDCID = 2 + // tgGroupCall 把 call 行转为 TL groupCall。 func tgGroupCall(call domain.GroupCall, viewerUserID int64, canManage bool) tg.GroupCallClass { if !call.Active() { @@ -39,6 +44,22 @@ func tgGroupCall(call domain.GroupCall, viewerUserID int64, canManage bool) tg.G UnmutedVideoLimit: groupCallUnmutedVideoLimit, Version: call.Version, } + if call.RtmpStream { + // RTMP 直播房间：观众经 broadcast 拉流。rtmp_stream 决定 TDesktop 打开 + // 直播 UI（而非语音聊天）；listeners_hidden 让观众数只用 participants_count + // 表达、不逐个下发 listener 行（RTMP 观众通常不进 participant 列表）。 + out.RtmpStream = true + out.ListenersHidden = true + // stream_dc_id 缺省会让 TDesktop fallback 到 main DC 并打 log；本服务单 DC， + // 显式回填本 DC id（拉流仍走同连接，DC shift 只影响客户端路由标记）。 + out.SetStreamDCID(groupCallStreamDCID) + } + if call.ScheduleDate > 0 { + // 定时通话：客户端据 schedule_date 显示倒计时面板；schedule_start_subscribed + // 是 per-viewer 投影（出站前由 applyScheduleSubscription 回填）。 + out.SetScheduleDate(call.ScheduleDate) + out.ScheduleStartSubscribed = call.ScheduleStartSubscribed + } if call.Title != "" { out.SetTitle(call.Title) } @@ -53,13 +74,19 @@ func tgGroupCall(call domain.GroupCall, viewerUserID int64, canManage bool) tg.G } // tgGroupCallParticipant 按 viewer 视角转换参与者行（Self flag per-viewer）。 +// join_as 频道身份输出 PeerChannel（客户端按 participant.peer==joinAs() 匹配自己， +// 输出用户 peer 会让以频道身份入会的本人面板出现 ghost 双行）。 func tgGroupCallParticipant(p domain.GroupCallParticipant, viewerUserID int64) tg.GroupCallParticipant { + var peer tg.PeerClass = &tg.PeerUser{UserID: p.UserID} + if p.JoinAsChannelID != 0 { + peer = &tg.PeerChannel{ChannelID: p.JoinAsChannelID} + } out := tg.GroupCallParticipant{ Muted: p.Muted, Left: p.Left, CanSelfUnmute: !p.MutedByAdmin, Self: p.UserID == viewerUserID, - Peer: &tg.PeerUser{UserID: p.UserID}, + Peer: peer, Date: p.JoinDate, Source: int(int32(uint32(p.SSRC))), // uint32 按位转 int32（join JSON 同款语义） } diff --git a/internal/rpc/deps.go b/internal/rpc/deps.go index bd3152f1..8a679ce1 100644 --- a/internal/rpc/deps.go +++ b/internal/rpc/deps.go @@ -705,6 +705,7 @@ type Deps struct { Polls PollsService Phone PhoneService GroupCalls GroupCallsService + LiveStreams LiveStreamsService SFU sfu.Service TURN turnsrv.Service LangPack LangPackService @@ -811,7 +812,13 @@ type PhoneService interface { // GroupCallsService 抽象超级群语音聊天信令（app/groupcalls）。 // 错误集合见 domain.ErrGroupCall*（rpc 层映射为 GROUPCALL_* RPC_ERROR）。 type GroupCallsService interface { - Create(ctx context.Context, channelID, creatorUserID int64, title string, now int) (domain.GroupCall, error) + Create(ctx context.Context, channelID, creatorUserID int64, title string, rtmpStream, joinMuted bool, scheduleDate, now int) (domain.GroupCall, error) + // RtmpStreamKey 取/轮换 channel 的持久 RTMP 推流密钥（rotate=true 覆盖旧 key）。 + RtmpStreamKey(ctx context.Context, channelID int64, rotate bool, now int) (string, error) + // StartScheduled / SetScheduleSubscription / ScheduleSubscriberIDs 是定时通话流程。 + StartScheduled(ctx context.Context, callID int64) (domain.GroupCall, bool, error) + SetScheduleSubscription(ctx context.Context, callID, userID int64, subscribed bool) error + ScheduleSubscriberIDs(ctx context.Context, callID int64) ([]int64, error) CreateConference(ctx context.Context, creatorUserID, randomID, migratedFromPhoneCallID int64, now int) (domain.GroupCall, error) Get(ctx context.Context, callID int64) (domain.GroupCall, bool, error) GetBySlug(ctx context.Context, slug string) (domain.GroupCall, bool, error) @@ -839,6 +846,17 @@ type GroupCallsService interface { ChainBlocks(ctx context.Context, callID int64, subChainID, offset, limit int) (domain.GroupCallChainBlockPage, error) } +// LiveStreamsService 抽象直播媒体面（app/livestream：RTMP ingest + 切段 ring）。 +// nil = 直播媒体面未启用（信令仍可用，观众停留在"等待推流"占位）。 +type LiveStreamsService interface { + // StreamChannels 返回 channel 当前直播时间轴；无活跃推流返回空。 + StreamChannels(channelID int64) []domain.LiveStreamChannel + // StreamPart 按 time_ms/scale 取一段打包好的 tgcalls broadcast part。 + StreamPart(channelID int64, timeMs int64, scale int) ([]byte, error) + // DropChannel 断开该 channel 的推流会话并清空缓冲（discard/revoke）。 + DropChannel(channelID int64) +} + // PollsService 抽象 poll 权威态的发送时创建与投票人列表（messages.getPollVotes）。 type PollsService interface { CreatePoll(ctx context.Context, def domain.PollDefinition) error diff --git a/internal/rpc/phone_group_call.go b/internal/rpc/phone_group_call.go index 5685455f..6a9cf701 100644 --- a/internal/rpc/phone_group_call.go +++ b/internal/rpc/phone_group_call.go @@ -116,6 +116,53 @@ func (r *Router) groupCallScopeFrom(ctx context.Context, in tg.InputGroupCallCla return &groupCallScope{userID: userID, call: call, channel: view.Channel, member: view.Self}, nil } +// groupCallJoinAsChannelID 解析 joinGroupCall.join_as：self（缺省/自己）→0； +// 本频道本身且 viewer 是 admin（匿名管理员/创建者语义，TDesktop RTMP createBox +// 对 creator 硬编码 joinAs=peer）→ channelID；其余身份返回 JOIN_AS_PEER_INVALID。 +func (r *Router) groupCallJoinAsChannelID(scope *groupCallScope, joinAs tg.InputPeerClass) (int64, error) { + switch v := joinAs.(type) { + case nil, *tg.InputPeerSelf, *tg.InputPeerEmpty: + return 0, nil + case *tg.InputPeerUser: + if v.UserID == scope.userID { + return 0, nil + } + case *tg.InputPeerChannel: + if !scope.call.Conference() && v.ChannelID == scope.channel.ID && channelMemberIsAdmin(scope.member) { + return v.ChannelID, nil + } + } + return 0, tgerr400("JOIN_AS_PEER_INVALID") +} + +// onPhoneGetGroupCallJoinAs 返回入会可选身份：所有人可用自己；频道 admin 额外 +// 可用频道本身（匿名身份）。TDesktop 在候选 >1 时显示 "join as" 选择框。 +func (r *Router) onPhoneGetGroupCallJoinAs(ctx context.Context, peer tg.InputPeerClass) (*tg.PhoneJoinAsPeers, error) { + userID, err := r.phoneRequireUser(ctx) + if err != nil { + return nil, err + } + out := &tg.PhoneJoinAsPeers{ + Peers: []tg.PeerClass{&tg.PeerUser{UserID: userID}}, + Chats: []tg.ChatClass{}, + Users: r.tgUsersForIDs(ctx, userID, []int64{userID}), + } + if r.deps.Channels == nil { + return out, nil + } + dp, err := r.checkedDomainPeerFromInputPeer(ctx, userID, peer) + if err != nil || dp.Type != domain.PeerTypeChannel || dp.ID == 0 { + return out, nil + } + view, err := r.deps.Channels.GetChannel(ctx, userID, dp.ID) + if err != nil || view.Self.Status != domain.ChannelMemberActive || !channelMemberIsAdmin(view.Self) { + return out, nil + } + out.Peers = append(out.Peers, &tg.PeerChannel{ChannelID: view.Channel.ID}) + out.Chats = append(out.Chats, tgChannel(userID, view.Channel, &view.Self)) + return out, nil +} + func (r *Router) conferenceCallCanAccess(ctx context.Context, callID, userID int64) (bool, error) { recipients, err := r.deps.GroupCalls.ConferenceRecipients(ctx, callID) if err != nil { @@ -140,11 +187,11 @@ func (r *Router) onPhoneCreateGroupCall(ctx context.Context, req *tg.PhoneCreate if err != nil { return nil, err } - if req.RtmpStream { - return nil, notImplementedErr() - } - if _, ok := req.GetScheduleDate(); ok { - return nil, notImplementedErr() + now := int(r.clock.Now().Unix()) + scheduleDate, _ := req.GetScheduleDate() + if scheduleDate < 0 || (scheduleDate > 0 && scheduleDate <= now) { + // TDesktop 选择器只给未来时间；过去时间直接拒绝（容忍在途秒差由客户端保证）。 + return nil, tgerr400("SCHEDULE_DATE_INVALID") } peer, err := r.checkedDomainPeerFromInputPeer(ctx, userID, req.Peer) if err != nil { @@ -160,22 +207,27 @@ func (r *Router) onPhoneCreateGroupCall(ctx context.Context, req *tg.PhoneCreate if view.Self.Status != domain.ChannelMemberActive || !channelMemberIsAdmin(view.Self) { return nil, tgerr400("CHAT_ADMIN_REQUIRED") } - if !view.Channel.Megagroup { - // broadcast 频道的 livestream 属范围外。 - return nil, notImplementedErr() - } - now := int(r.clock.Now().Unix()) - call, err := r.deps.GroupCalls.Create(ctx, view.Channel.ID, userID, req.Title, now) + // 广播频道直播、以及任意 RTMP 直播：参与者都是纯观众（listener），入会即 + // 强制静音且不可自解（join_muted）。RTMP 尤其关键——TDesktop 在 stream 模式下 + // 若发现 self 行非 force-muted 会每 3s `Rejoin after unforcemute`，导致死循环。 + joinMuted := !view.Channel.Megagroup || req.RtmpStream + call, err := r.deps.GroupCalls.Create(ctx, view.Channel.ID, userID, req.Title, req.RtmpStream, joinMuted, scheduleDate, now) if err != nil { return nil, groupCallErr(err) } - // started 服务消息（带频道 pts，离线成员经 channels difference 补收）。 - var serviceRes domain.SendChannelMessageResult - if res, err := r.deps.Channels.AppendCallServiceMessage(ctx, view.Channel.ID, userID, now, domain.ChannelMessageAction{ + // 服务消息（带频道 pts，离线成员经 channels difference 补收）： + // 定时通话发 scheduled（"预约了视频聊天"），立即通话发 started。 + serviceAction := domain.ChannelMessageAction{ Type: domain.ChannelActionGroupCall, CallID: call.ID, CallAccessHash: call.AccessHash, - }); err == nil { + } + if scheduleDate > 0 { + serviceAction.Type = domain.ChannelActionGroupCallScheduled + serviceAction.CallScheduleDate = scheduleDate + } + var serviceRes domain.SendChannelMessageResult + if res, err := r.deps.Channels.AppendCallServiceMessage(ctx, view.Channel.ID, userID, now, serviceAction); err == nil { serviceRes = res _ = r.deps.GroupCalls.SetStartedMessageID(ctx, call.ID, res.Message.ID) } else { @@ -216,6 +268,13 @@ func (r *Router) onPhoneJoinGroupCall(ctx context.Context, req *tg.PhoneJoinGrou if !scope.call.Active() { return nil, groupCallAlreadyDiscardedErr() } + if scope.call.RtmpStream { + return r.joinRtmpGroupCall(ctx, scope, req) + } + joinAsChannelID, err := r.groupCallJoinAsChannelID(scope, req.JoinAs) + if err != nil { + return nil, err + } // 解析上行 join JSON（容忍 video_stopped 等 flag 与 ssrc-groups——TDesktop join 即带）。 offer, ssrc, err := parseGroupCallJoinPayload(req.Params.Data) if err != nil { @@ -247,15 +306,16 @@ func (r *Router) onPhoneJoinGroupCall(ctx context.Context, req *tg.PhoneJoinGrou Active: !req.VideoStopped && len(offer.SsrcGroups) > 0, } mut, err := r.deps.GroupCalls.Join(ctx, domain.JoinGroupCallRequest{ - CallID: scope.call.ID, - UserID: scope.userID, - SSRC: ssrc, - Muted: req.Muted, - IsAdmin: scope.canManage(), - PublicKey: publicKey, - JoinBlock: joinBlock, - VideoJSON: encodeVideoState(videoState), - Now: now, + CallID: scope.call.ID, + UserID: scope.userID, + JoinAsChannelID: joinAsChannelID, + SSRC: ssrc, + Muted: req.Muted, + IsAdmin: scope.canManage(), + PublicKey: publicKey, + JoinBlock: joinBlock, + VideoJSON: encodeVideoState(videoState), + Now: now, }) if err != nil { return nil, groupCallErr(err) @@ -377,6 +437,10 @@ func (r *Router) onPhoneDiscardGroupCall(ctx context.Context, in tg.InputGroupCa return r.groupCallUpdateContainer(ctx, scope.userID, domain.Channel{}, groupCallUpdateFor(domain.Channel{}, call, scope.userID, true), nil), nil } + // RTMP 直播结束：断开推流并清空缓冲（观众后续拉流转 resync/停止）。 + if call.RtmpStream && r.deps.LiveStreams != nil { + r.deps.LiveStreams.DropChannel(scope.channel.ID) + } // 清 channel 关联 + ended 服务消息（带 duration）。 channel := scope.channel if updated, err := r.deps.Channels.SetActiveCall(ctx, channel.ID, 0, 0, false); err == nil { @@ -432,8 +496,10 @@ func (r *Router) onPhoneGetGroupCall(ctx context.Context, req *tg.PhoneGetGroupC if !scope.call.Conference() { chats = append(chats, tgChannel(scope.userID, scope.channel, &scope.member)) } + // 定时通话：回填 viewer 自己的开播提醒订阅（客户端 reload 全量重建本地状态）。 + call := r.applyScheduleSubscription(ctx, scope.call, scope.userID) return &tg.PhoneGroupCall{ - Call: tgGroupCall(scope.call, scope.userID, scope.canManage()), + Call: tgGroupCall(call, scope.userID, scope.canManage()), Participants: tgGroupCallParticipants(page.Participants, scope.userID), ParticipantsNextOffset: page.NextOffset, Chats: chats, diff --git a/internal/rpc/phone_group_call_joinas_test.go b/internal/rpc/phone_group_call_joinas_test.go new file mode 100644 index 00000000..1fcac73f --- /dev/null +++ b/internal/rpc/phone_group_call_joinas_test.go @@ -0,0 +1,113 @@ +package rpc + +import ( + "testing" + + "github.com/gotd/td/tg" +) + +// TestGroupCallJoinAsChannel 覆盖 join_as 身份闭环：admin 以频道身份入会 → +// 参与者行 peer=PeerChannel（本人与对端视角一致）；非 admin 以频道身份被拒； +// getGroupCallJoinAs 对 admin 返回 self+频道两个候选、普通成员只返回 self。 +func TestGroupCallJoinAsChannel(t *testing.T) { + f := newGroupCallFixture(t) + ownerCtx := f.userCtx(f.owner, 11) + memberCtx := f.userCtx(f.member, 22) + channelPeer := &tg.InputPeerChannel{ChannelID: f.channel.ID, AccessHash: f.channel.AccessHash} + + // --- getGroupCallJoinAs 候选 --- + ownerJoinAs, err := f.router.onPhoneGetGroupCallJoinAs(ownerCtx, channelPeer) + if err != nil { + t.Fatalf("owner getGroupCallJoinAs: %v", err) + } + if len(ownerJoinAs.Peers) != 2 { + t.Fatalf("owner join-as peers = %d, want 2 (self + channel): %+v", len(ownerJoinAs.Peers), ownerJoinAs.Peers) + } + if _, ok := ownerJoinAs.Peers[1].(*tg.PeerChannel); !ok { + t.Fatalf("owner join-as second peer = %T, want PeerChannel", ownerJoinAs.Peers[1]) + } + memberJoinAs, err := f.router.onPhoneGetGroupCallJoinAs(memberCtx, channelPeer) + if err != nil { + t.Fatalf("member getGroupCallJoinAs: %v", err) + } + if len(memberJoinAs.Peers) != 1 { + t.Fatalf("member join-as peers = %d, want 1 (self only)", len(memberJoinAs.Peers)) + } + + // --- create + owner 以频道身份 join --- + createRes, err := f.router.onPhoneCreateGroupCall(ownerCtx, &tg.PhoneCreateGroupCallRequest{ + Peer: channelPeer, RandomID: 1, + }) + if err != nil { + t.Fatalf("createGroupCall: %v", err) + } + call := findUpdate[*tg.UpdateGroupCall](t, createRes).Call.(*tg.GroupCall) + input := &tg.InputGroupCall{ID: call.ID, AccessHash: call.AccessHash} + + joinRes, err := f.router.onPhoneJoinGroupCall(ownerCtx, &tg.PhoneJoinGroupCallRequest{ + Call: input, + JoinAs: channelPeer, + Params: groupCallJoinParams(t, 8001), + }) + if err != nil { + t.Fatalf("owner joinGroupCall(join_as=channel): %v", err) + } + participants := findUpdate[*tg.UpdateGroupCallParticipants](t, joinRes) + if len(participants.Participants) != 1 { + t.Fatalf("participants = %d, want 1", len(participants.Participants)) + } + self := participants.Participants[0] + peerCh, ok := self.Peer.(*tg.PeerChannel) + if !ok || peerCh.ChannelID != f.channel.ID { + t.Fatalf("self participant peer = %#v, want PeerChannel(%d)", self.Peer, f.channel.ID) + } + if !self.Self { + t.Fatalf("join_as channel row missing self flag for the joining user") + } + + // --- 对端视角：member 拉参与者列表也看到频道身份 --- + page, err := f.router.onPhoneGetGroupParticipants(memberCtx, &tg.PhoneGetGroupParticipantsRequest{ + Call: input, Limit: 10, + }) + if err != nil { + t.Fatalf("member getGroupParticipants: %v", err) + } + if len(page.Participants) != 1 { + t.Fatalf("member sees %d participants, want 1", len(page.Participants)) + } + if pc, ok := page.Participants[0].Peer.(*tg.PeerChannel); !ok || pc.ChannelID != f.channel.ID { + t.Fatalf("member view participant peer = %#v, want PeerChannel(%d)", page.Participants[0].Peer, f.channel.ID) + } + if page.Participants[0].Self { + t.Fatalf("member view incorrectly flags channel row as self") + } + + // --- 非 admin 以频道身份 join 被拒 --- + if _, err := f.router.onPhoneJoinGroupCall(memberCtx, &tg.PhoneJoinGroupCallRequest{ + Call: input, + JoinAs: channelPeer, + Params: groupCallJoinParams(t, 8002), + }); err == nil { + t.Fatalf("non-admin joined as channel") + } else { + assertPhoneRPCErr(t, err, "JOIN_AS_PEER_INVALID") + } + + // --- rejoin 换回本人身份：行替换而非新增 --- + rejoinRes, err := f.router.onPhoneJoinGroupCall(ownerCtx, &tg.PhoneJoinGroupCallRequest{ + Call: input, + JoinAs: &tg.InputPeerSelf{}, + Params: groupCallJoinParams(t, 8003), + }) + if err != nil { + t.Fatalf("owner rejoin as self: %v", err) + } + rejoined := findUpdate[*tg.UpdateGroupCallParticipants](t, rejoinRes).Participants[0] + if _, ok := rejoined.Peer.(*tg.PeerUser); !ok { + t.Fatalf("rejoin-as-self participant peer = %#v, want PeerUser", rejoined.Peer) + } + page2, _ := f.router.onPhoneGetGroupParticipants(memberCtx, &tg.PhoneGetGroupParticipantsRequest{Call: input, Limit: 10}) + if len(page2.Participants) != 1 { + t.Fatalf("after identity switch participants = %d, want 1 (replace not add)", len(page2.Participants)) + } +} diff --git a/internal/rpc/phone_group_call_rtmp.go b/internal/rpc/phone_group_call_rtmp.go new file mode 100644 index 00000000..0ee2895a --- /dev/null +++ b/internal/rpc/phone_group_call_rtmp.go @@ -0,0 +1,192 @@ +package rpc + +import ( + "context" + "crypto/rand" + "encoding/binary" + "encoding/json" + + "github.com/gotd/td/tg" + + "telesrv/internal/domain" +) + +// RTMP 直播（Live Stream）RPC：createGroupCall(rtmp_stream) 建房后，推流方（OBS） +// 用 getGroupCallStreamRtmpUrl 拿到的 url/key 推流，观众 join 后经 +// upload.getFile(inputGroupCallStream) 拉 broadcast chunk。 +// +// 与普通语音聊天（RTC/SFU）关键差异：RTMP join 不建 SFU 连接、不做 ssrc 唯一性 +// 媒体面校验；updateGroupCallConnection.params 返回 {"stream":true,"rtmp":true} +// 让 TDesktop 切 broadcast 模式（ParseJoinResponse → JoinBroadcastStream）。 + +// rtmpConnectionParams 是 RTMP 观众 join 响应的下行 JSON（tgcalls broadcast 分支）。 +// 可管理者附 rtmp_stream_url/key 供直播设置页展示（普通观众不下发，避免泄漏推流凭据）。 +type rtmpConnectionParams struct { + Stream bool `json:"stream"` + Rtmp bool `json:"rtmp"` + RtmpStreamURL string `json:"rtmp_stream_url,omitempty"` + RtmpStreamKey string `json:"rtmp_stream_key,omitempty"` +} + +func buildRtmpConnectionParams(url, key string) (string, error) { + p := rtmpConnectionParams{Stream: true, Rtmp: true, RtmpStreamURL: url, RtmpStreamKey: key} + out, err := json.Marshal(p) + if err != nil { + return "", err + } + return string(out), nil +} + +// joinRtmpGroupCall 处理 RTMP 直播房间的 joinGroupCall。 +func (r *Router) joinRtmpGroupCall(ctx context.Context, scope *groupCallScope, req *tg.PhoneJoinGroupCallRequest) (tg.UpdatesClass, error) { + // 房间上限：RTMP 观众也计入 participants_count；rejoin（已在会换 ssrc）不受限。 + if max := r.cfg.GroupCallMaxParticipants; max > 0 && scope.call.ParticipantsCount >= max { + if p, found, _ := r.deps.GroupCalls.Participant(ctx, scope.call.ID, scope.userID); !found || p.Left { + return nil, groupCallForbiddenErr() + } + } + joinAsChannelID, err := r.groupCallJoinAsChannelID(scope, req.JoinAs) + if err != nil { + return nil, err + } + now := int(r.clock.Now().Unix()) + // RTMP 观众的 join JSON 仍带 tgcalls ssrc（客户端为拉流也建了本地 controller）； + // 解析失败/缺失不致命——直播不需要 SFU，用随机 ssrc 记账保证参与者行有效。 + ssrc := int64(0) + if _, s, err := parseGroupCallJoinPayload(req.Params.Data); err == nil { + ssrc = s + } + if ssrc == 0 { + ssrc = randomSSRC() + } + // RTMP 观众（含创建者，其经 OBS 独立推流，在 group call 里同样是纯观众）一律 + // force-muted：IsAdmin=false 让 store 对 join_muted 房间置 muted+muted_by_admin， + // self 行输出 muted=true / can_self_unmute=false，TDesktop 才会稳定停在 stream 模式。 + mut, err := r.deps.GroupCalls.Join(ctx, domain.JoinGroupCallRequest{ + CallID: scope.call.ID, + UserID: scope.userID, + JoinAsChannelID: joinAsChannelID, + SSRC: ssrc, + Muted: true, + IsAdmin: false, + Now: now, + }) + if err != nil { + return nil, groupCallErr(err) + } + // 可管理者拿推流 url/key（直播设置页展示）；普通观众只得 stream:true。 + var url, key string + if scope.canManage() && r.deps.GroupCalls != nil { + if k, kerr := r.deps.GroupCalls.RtmpStreamKey(ctx, scope.channel.ID, false, now); kerr == nil { + key = k + url = r.rtmpIngestURL() + } + } + params, err := buildRtmpConnectionParams(url, key) + if err != nil { + return nil, internalErr() + } + channel := r.groupCallMutationFanout(ctx, scope.channel, mut) + out := r.groupCallUpdateContainer(ctx, scope.userID, channel, + &tg.UpdateGroupCallParticipants{ + Call: &tg.InputGroupCall{ID: mut.Call.ID, AccessHash: mut.Call.AccessHash}, + Participants: tgGroupCallParticipants([]domain.GroupCallParticipant{mut.Participant}, scope.userID), + Version: mut.Call.Version, + }, []int64{scope.userID}) + // updateGroupCall 必须先于 updateGroupCallConnection：TDesktop 按序 applyUpdates， + // 处理 connection 时若还没从 groupCall 读到 stream_dc_id 会打 + // "Api Error: Empty stream_dc_id" 并 fallback 主 DC。 + callUpdate := &tg.UpdateGroupCall{Call: tgGroupCall(mut.Call, scope.userID, scope.canManage())} + if channel.ID != 0 { + callUpdate.SetPeer(&tg.PeerChannel{ChannelID: channel.ID}) + } + out.Updates = append(out.Updates, callUpdate) + out.Updates = append(out.Updates, &tg.UpdateGroupCallConnection{Params: tg.DataJSON{Data: params}}) + return out, nil +} + +// onPhoneGetGroupCallStreamRtmpURL 返回频道的 RTMP 推流 url/key（创建前预览、 +// 直播设置页展示、revoke 轮换）。仅频道管理员可调；revoke=true 生成新 key，旧 key +// 立即失效并断开正在进行的推流。 +func (r *Router) onPhoneGetGroupCallStreamRtmpURL(ctx context.Context, req *tg.PhoneGetGroupCallStreamRtmpURLRequest) (*tg.PhoneGroupCallStreamRtmpURL, error) { + if req == nil { + return nil, inputRequestInvalidErr() + } + if r.deps.GroupCalls == nil || r.deps.Channels == nil { + return nil, notImplementedErr() + } + userID, err := r.phoneRequireUser(ctx) + if err != nil { + return nil, err + } + peer, err := r.checkedDomainPeerFromInputPeer(ctx, userID, req.Peer) + if err != nil { + return nil, err + } + if peer.Type != domain.PeerTypeChannel || peer.ID == 0 { + return nil, peerIDInvalidErr() + } + view, err := r.deps.Channels.GetChannel(ctx, userID, peer.ID) + if err != nil { + return nil, peerIDInvalidErr() + } + if view.Self.Status != domain.ChannelMemberActive || !channelMemberIsAdmin(view.Self) { + return nil, tgerr400("CHAT_ADMIN_REQUIRED") + } + now := int(r.clock.Now().Unix()) + key, err := r.deps.GroupCalls.RtmpStreamKey(ctx, peer.ID, req.Revoke, now) + if err != nil { + return nil, groupCallErr(err) + } + if req.Revoke && r.deps.LiveStreams != nil { + // 旧 key 失效后仍在推的连接必须断开（否则用旧 key 的推流会继续被接收）。 + r.deps.LiveStreams.DropChannel(peer.ID) + } + return &tg.PhoneGroupCallStreamRtmpURL{URL: r.rtmpIngestURL(), Key: key}, nil +} + +// onPhoneGetGroupCallStreamChannels 返回 RTMP 直播的当前时间轴（unified：channel=1 +// scale=0），供 tgcalls 决定从哪个 time_ms 起拉 chunk。无活跃推流时返回空列表， +// TDesktop 据此显示"等待推流"占位并循环重试。 +func (r *Router) onPhoneGetGroupCallStreamChannels(ctx context.Context, call tg.InputGroupCallClass) (*tg.PhoneGroupCallStreamChannels, error) { + scope, err := r.groupCallScopeFrom(ctx, call) + if err != nil { + return nil, err + } + out := &tg.PhoneGroupCallStreamChannels{Channels: []tg.GroupCallStreamChannel{}} + if !scope.call.RtmpStream || r.deps.LiveStreams == nil || scope.channel.ID == 0 { + return out, nil + } + for _, ch := range r.deps.LiveStreams.StreamChannels(scope.channel.ID) { + out.Channels = append(out.Channels, tg.GroupCallStreamChannel{ + Channel: ch.Channel, + Scale: ch.Scale, + LastTimestampMs: ch.LastTimestampMs, + }) + } + return out, nil +} + +// rtmpIngestURL 返回展示给推流端的 RTMP 服务器地址。 +func (r *Router) rtmpIngestURL() string { + if r.cfg.RtmpIngestURL != "" { + return r.cfg.RtmpIngestURL + } + host := r.cfg.IP + if host == "" { + host = "127.0.0.1" + } + return "rtmp://" + host + ":2400/live" +} + +func randomSSRC() int64 { + var buf [4]byte + if _, err := rand.Read(buf[:]); err != nil { + return 1 + } + v := int64(binary.BigEndian.Uint32(buf[:])) + if v == 0 { + v = 1 + } + return v +} diff --git a/internal/rpc/phone_group_call_rtmp_test.go b/internal/rpc/phone_group_call_rtmp_test.go new file mode 100644 index 00000000..44573b11 --- /dev/null +++ b/internal/rpc/phone_group_call_rtmp_test.go @@ -0,0 +1,267 @@ +package rpc + +import ( + "encoding/json" + "testing" + "time" + + "github.com/gotd/td/tg" + "go.uber.org/zap/zaptest" + + appchannels "telesrv/internal/app/channels" + appgroupcalls "telesrv/internal/app/groupcalls" + appusers "telesrv/internal/app/users" + "telesrv/internal/domain" + "telesrv/internal/store/memory" +) + +// fakeLiveStreams 是 LiveStreamsService 的测试替身，按 channelID 存可拉流段。 +type fakeLiveStreams struct { + channels map[int64][]domain.LiveStreamChannel + parts map[int64]map[int64][]byte // channelID → time_ms → part + dropped map[int64]bool +} + +func newFakeLiveStreams() *fakeLiveStreams { + return &fakeLiveStreams{ + channels: map[int64][]domain.LiveStreamChannel{}, + parts: map[int64]map[int64][]byte{}, + dropped: map[int64]bool{}, + } +} + +func (f *fakeLiveStreams) StreamChannels(channelID int64) []domain.LiveStreamChannel { + return f.channels[channelID] +} + +func (f *fakeLiveStreams) StreamPart(channelID int64, timeMs int64, scale int) ([]byte, error) { + if scale != 0 { + return nil, domain.ErrLiveStreamPartExpired + } + byTime, ok := f.parts[channelID] + if !ok { + return nil, domain.ErrLiveStreamNoStream + } + part, ok := byTime[timeMs] + if !ok { + return nil, domain.ErrLiveStreamPartNotReady + } + return part, nil +} + +func (f *fakeLiveStreams) DropChannel(channelID int64) { f.dropped[channelID] = true } + +type rtmpFixture struct { + *groupCallFixture + live *fakeLiveStreams +} + +// newRtmpFixture 复制 newGroupCallFixture 的用户/频道搭建，但注入 LiveStreams 替身。 +func newRtmpFixture(t *testing.T) *rtmpFixture { + t.Helper() + ctx := t.Context() + userStore := memory.NewUserStore() + channelStore := memory.NewChannelStore() + sessions := &groupCallSessions{} + clk := &phoneTestClock{now: time.Unix(1_700_000_000, 0)} + live := newFakeLiveStreams() + router := New(Config{GroupCallMaxParticipants: 8, IP: "203.0.113.7"}, Deps{ + Users: appusers.NewService(userStore), + Channels: appchannels.NewService(channelStore), + GroupCalls: appgroupcalls.NewService(memory.NewGroupCallStore()), + LiveStreams: live, + Sessions: sessions, + }, zaptest.NewLogger(t), clk) + f := &groupCallFixture{t: t, ctx: ctx, router: router, sessions: sessions, clk: clk} + mk := func(hash int64, phone, name string) domain.User { + u, err := userStore.Create(ctx, domain.User{AccessHash: hash, Phone: phone, FirstName: name}) + if err != nil { + t.Fatalf("create user %s: %v", name, err) + } + return u + } + f.owner = mk(2001, "13900000001", "Owner") + f.member = mk(2002, "13900000002", "Member") + f.outsider = mk(2003, "13900000003", "Outsider") + created, err := router.onMessagesCreateChat(f.userCtx(f.owner, 11), &tg.MessagesCreateChatRequest{ + Users: []tg.InputUserClass{&tg.InputUser{UserID: f.member.ID, AccessHash: f.member.AccessHash}}, + Title: "live room", + }) + if err != nil { + t.Fatalf("create chat: %v", err) + } + for _, chat := range created.Updates.(*tg.Updates).Chats { + if ch, ok := chat.(*tg.Channel); ok { + f.channel = ch + break + } + } + if f.channel == nil { + t.Fatalf("no channel in create chat result") + } + f.sessions.online = []int64{f.owner.ID, f.member.ID} + f.sessions.reset() + return &rtmpFixture{groupCallFixture: f, live: live} +} + +func (f *rtmpFixture) createLive(t *testing.T) *tg.GroupCall { + t.Helper() + res, err := f.router.onPhoneCreateGroupCall(f.userCtx(f.owner, 11), &tg.PhoneCreateGroupCallRequest{ + Peer: &tg.InputPeerChannel{ChannelID: f.channel.ID, AccessHash: f.channel.AccessHash}, + RandomID: 1, + RtmpStream: true, + }) + if err != nil { + t.Fatalf("createGroupCall(rtmp): %v", err) + } + call := findUpdate[*tg.UpdateGroupCall](t, res).Call.(*tg.GroupCall) + if !call.RtmpStream { + t.Fatalf("created group call missing rtmp_stream flag: %+v", call) + } + if _, ok := call.GetStreamDCID(); !ok { + t.Fatalf("rtmp group call missing stream_dc_id") + } + return call +} + +// TestRtmpCreateAndAdminGetUrl 验证 RTMP 直播创建后管理员可取 url/key，revoke 轮换 key。 +func TestRtmpCreateAndAdminGetUrl(t *testing.T) { + f := newRtmpFixture(t) + ownerCtx := f.userCtx(f.owner, 11) + f.createLive(t) + + res, err := f.router.onPhoneGetGroupCallStreamRtmpURL(ownerCtx, &tg.PhoneGetGroupCallStreamRtmpURLRequest{ + Peer: &tg.InputPeerChannel{ChannelID: f.channel.ID, AccessHash: f.channel.AccessHash}, + }) + if err != nil { + t.Fatalf("getGroupCallStreamRtmpUrl: %v", err) + } + if res.URL == "" || res.Key == "" { + t.Fatalf("empty rtmp url/key: %+v", res) + } + key1 := res.Key + + // 非管理员不得取推流凭据。 + if _, err := f.router.onPhoneGetGroupCallStreamRtmpURL(f.userCtx(f.member, 22), &tg.PhoneGetGroupCallStreamRtmpURLRequest{ + Peer: &tg.InputPeerChannel{ChannelID: f.channel.ID, AccessHash: f.channel.AccessHash}, + }); err == nil { + t.Fatalf("non-admin got rtmp url without error") + } + + // revoke 轮换 key 并断开推流会话。 + res2, err := f.router.onPhoneGetGroupCallStreamRtmpURL(ownerCtx, &tg.PhoneGetGroupCallStreamRtmpURLRequest{ + Peer: &tg.InputPeerChannel{ChannelID: f.channel.ID, AccessHash: f.channel.AccessHash}, + Revoke: true, + }) + if err != nil { + t.Fatalf("getGroupCallStreamRtmpUrl(revoke): %v", err) + } + if res2.Key == key1 { + t.Fatalf("revoke did not rotate key") + } + if !f.live.dropped[f.channel.ID] { + t.Fatalf("revoke did not drop live stream channel") + } +} + +// TestRtmpJoinReturnsStreamParams 验证 RTMP viewer join 返回 stream:true 的 connection params， +// 管理员附带 rtmp_stream_url/key，普通观众不下发凭据。 +func TestRtmpJoinReturnsStreamParams(t *testing.T) { + f := newRtmpFixture(t) + call := f.createLive(t) + + // 管理员 join：带 url/key。 + ownerRes, err := f.router.onPhoneJoinGroupCall(f.userCtx(f.owner, 12), &tg.PhoneJoinGroupCallRequest{ + Call: &tg.InputGroupCall{ID: call.ID, AccessHash: call.AccessHash}, + JoinAs: &tg.InputPeerSelf{}, + Params: groupCallJoinParams(t, 5001), + }) + if err != nil { + t.Fatalf("owner joinGroupCall(rtmp): %v", err) + } + conn := findUpdate[*tg.UpdateGroupCallConnection](t, ownerRes) + var params rtmpConnectionParams + if err := json.Unmarshal([]byte(conn.Params.Data), ¶ms); err != nil { + t.Fatalf("parse connection params: %v", err) + } + if !params.Stream || !params.Rtmp { + t.Fatalf("owner connection params not stream/rtmp: %+v", params) + } + if params.RtmpStreamURL == "" || params.RtmpStreamKey == "" { + t.Fatalf("admin join missing rtmp url/key: %+v", params) + } + + // 普通成员 join：stream:true 但无凭据。 + memberRes, err := f.router.onPhoneJoinGroupCall(f.userCtx(f.member, 22), &tg.PhoneJoinGroupCallRequest{ + Call: &tg.InputGroupCall{ID: call.ID, AccessHash: call.AccessHash}, + JoinAs: &tg.InputPeerSelf{}, + Params: groupCallJoinParams(t, 5002), + }) + if err != nil { + t.Fatalf("member joinGroupCall(rtmp): %v", err) + } + connM := findUpdate[*tg.UpdateGroupCallConnection](t, memberRes) + var paramsM rtmpConnectionParams + if err := json.Unmarshal([]byte(connM.Params.Data), ¶msM); err != nil { + t.Fatalf("parse member connection params: %v", err) + } + if !paramsM.Stream || !paramsM.Rtmp { + t.Fatalf("member connection params not stream/rtmp: %+v", paramsM) + } + if paramsM.RtmpStreamURL != "" || paramsM.RtmpStreamKey != "" { + t.Fatalf("non-admin join leaked rtmp url/key: %+v", paramsM) + } +} + +// TestRtmpGetStreamPart 验证 upload.getFile(inputGroupCallStream) 的取段与错误映射。 +func TestRtmpGetStreamPart(t *testing.T) { + f := newRtmpFixture(t) + call := f.createLive(t) + memberCtx := f.userCtx(f.member, 22) + // 观众须先 join（拉流校验 join 状态经 scope）。 + if _, err := f.router.onPhoneJoinGroupCall(memberCtx, &tg.PhoneJoinGroupCallRequest{ + Call: &tg.InputGroupCall{ID: call.ID, AccessHash: call.AccessHash}, + JoinAs: &tg.InputPeerSelf{}, + Params: groupCallJoinParams(t, 6001), + }); err != nil { + t.Fatalf("member join: %v", err) + } + // 备好一段可拉数据。 + f.live.channels[f.channel.ID] = []domain.LiveStreamChannel{{Channel: 1, Scale: 0, LastTimestampMs: 3000}} + f.live.parts[f.channel.ID] = map[int64][]byte{3000: []byte("SEGMENT-BYTES-0123456789")} + + loc := func(timeMs int64) *tg.InputGroupCallStream { + return &tg.InputGroupCallStream{Call: &tg.InputGroupCall{ID: call.ID, AccessHash: call.AccessHash}, TimeMs: timeMs, Scale: 0} + } + + // 命中：分片切片。 + out, err := f.router.onUploadGetFile(memberCtx, &tg.UploadGetFileRequest{Location: loc(3000), Offset: 0, Limit: 8}) + if err != nil { + t.Fatalf("getFile stream: %v", err) + } + uf := out.(*tg.UploadFile) + if string(uf.Bytes) != "SEGMENT-" { + t.Fatalf("stream chunk = %q, want first 8 bytes", uf.Bytes) + } + // 续段（offset 中段）。 + out2, _ := f.router.onUploadGetFile(memberCtx, &tg.UploadGetFileRequest{Location: loc(3000), Offset: 8, Limit: 1 << 17}) + if string(out2.(*tg.UploadFile).Bytes) != "BYTES-0123456789" { + t.Fatalf("stream chunk tail = %q", out2.(*tg.UploadFile).Bytes) + } + + // 未就绪段 → TIME_TOO_BIG。 + if _, err := f.router.onUploadGetFile(memberCtx, &tg.UploadGetFileRequest{Location: loc(4000), Offset: 0, Limit: 1024}); err == nil { + t.Fatalf("expected TIME_TOO_BIG for future segment") + } else { + assertPhoneRPCErr(t, err, "TIME_TOO_BIG") + } + + // getGroupCallStreamChannels 返回时间轴。 + chRes, err := f.router.onPhoneGetGroupCallStreamChannels(memberCtx, &tg.InputGroupCall{ID: call.ID, AccessHash: call.AccessHash}) + if err != nil { + t.Fatalf("getGroupCallStreamChannels: %v", err) + } + if len(chRes.Channels) != 1 || chRes.Channels[0].LastTimestampMs != 3000 { + t.Fatalf("stream channels = %+v", chRes.Channels) + } +} diff --git a/internal/rpc/phone_group_call_scheduled.go b/internal/rpc/phone_group_call_scheduled.go new file mode 100644 index 00000000..a76706ee --- /dev/null +++ b/internal/rpc/phone_group_call_scheduled.go @@ -0,0 +1,112 @@ +package rpc + +import ( + "context" + + "github.com/gotd/td/tg" + "go.uber.org/zap" + + "telesrv/internal/domain" +) + +// Scheduled video chat（定时通话）RPC。客户端流程（TDesktop calls_group_call.cpp）： +// createGroupCall(schedule_date) → State::Waiting 倒计时（不 join）→ 管理员 +// startScheduledGroupCall 清 schedule_date → updateGroupCall（无 schedule_date）→ +// 客户端 setScheduledDate(0) 触发 initialJoin 正式入会。开播提醒是 per-viewer 的 +// schedule_start_subscribed flag（toggleGroupCallStartSubscription）。 + +// applyScheduleSubscription 为单个 viewer 回填 ScheduleStartSubscribed 投影字段。 +func (r *Router) applyScheduleSubscription(ctx context.Context, call domain.GroupCall, viewerUserID int64) domain.GroupCall { + if call.ScheduleDate == 0 || viewerUserID == 0 { + return call + } + subs, err := r.deps.GroupCalls.ScheduleSubscriberIDs(ctx, call.ID) + if err != nil { + return call + } + for _, id := range subs { + if id == viewerUserID { + call.ScheduleStartSubscribed = true + break + } + } + return call +} + +func (r *Router) onPhoneStartScheduledGroupCall(ctx context.Context, in tg.InputGroupCallClass) (tg.UpdatesClass, error) { + scope, err := r.groupCallScopeFrom(ctx, in) + if err != nil { + return nil, err + } + if scope.call.Conference() || scope.channel.ID == 0 { + return nil, groupCallInvalidErr() + } + if !scope.canManage() { + return nil, tgerr400("CHAT_ADMIN_REQUIRED") + } + call, changed, err := r.deps.GroupCalls.StartScheduled(ctx, scope.call.ID) + if err != nil { + return nil, groupCallErr(err) + } + channel := scope.channel + if !changed { + // 幂等：已开始，只回快照，不重复扇出/服务消息。 + return r.groupCallUpdateContainer(ctx, scope.userID, channel, + groupCallUpdateFor(channel, call, scope.userID, true), nil), nil + } + now := int(r.clock.Now().Unix()) + // started 服务消息（与即时创建的 started 同构）。 + var serviceRes domain.SendChannelMessageResult + if res, err := r.deps.Channels.AppendCallServiceMessage(ctx, channel.ID, scope.userID, now, domain.ChannelMessageAction{ + Type: domain.ChannelActionGroupCall, + CallID: call.ID, + CallAccessHash: call.AccessHash, + }); err == nil { + serviceRes = res + _ = r.deps.GroupCalls.SetStartedMessageID(ctx, call.ID, res.Message.ID) + } else { + r.log.Warn("scheduled group call started service message", zap.Int64("channel_id", channel.ID), zap.Error(err)) + } + // 扇出：updateGroupCall（schedule_date 已清，客户端据此自动入会）+ 服务消息。 + // 订阅者与普通在线成员走同一在线扇出；离线订阅者的推送提醒（push notification） + // 属通知系统范围，当前不实现（记矩阵 todo）。 + r.pushGroupCallUpdate(ctx, channel, call) + if serviceRes.Event.Pts != 0 { + r.pushGroupCallServiceMessage(ctx, scope.userID, serviceRes) + } + out := r.groupCallUpdateContainer(ctx, scope.userID, channel, + groupCallUpdateFor(channel, call, scope.userID, true), nil) + if serviceRes.Event.Pts != 0 { + if msgUpdate := tgChannelUpdate(scope.userID, serviceRes.Event); msgUpdate != nil { + out.Updates = append(out.Updates, msgUpdate) + } + } + return out, nil +} + +func (r *Router) onPhoneToggleGroupCallStartSubscription(ctx context.Context, req *tg.PhoneToggleGroupCallStartSubscriptionRequest) (tg.UpdatesClass, error) { + if req == nil { + return nil, inputRequestInvalidErr() + } + scope, err := r.groupCallScopeFrom(ctx, req.Call) + if err != nil { + return nil, err + } + if !scope.call.Active() { + return nil, groupCallAlreadyDiscardedErr() + } + if scope.call.ScheduleDate == 0 { + // 只有未开始的定时通话才有开播提醒可订。 + return nil, groupCallInvalidErr() + } + if err := r.deps.GroupCalls.SetScheduleSubscription(ctx, scope.call.ID, scope.userID, req.Subscribed); err != nil { + return nil, groupCallErr(err) + } + call := scope.call + call.ScheduleStartSubscribed = req.Subscribed + // 订阅是 per-viewer 私有状态：响应给本设备，推送同步本人其它在线设备即可。 + update := groupCallUpdateFor(scope.channel, call, scope.userID, scope.canManage()) + r.pushUserMessage(ctx, scope.userID, "schedule subscription update", + r.groupCallUpdateContainer(ctx, scope.userID, scope.channel, update, nil)) + return r.groupCallUpdateContainer(ctx, scope.userID, scope.channel, update, nil), nil +} diff --git a/internal/rpc/phone_group_call_scheduled_test.go b/internal/rpc/phone_group_call_scheduled_test.go new file mode 100644 index 00000000..d3bb7afc --- /dev/null +++ b/internal/rpc/phone_group_call_scheduled_test.go @@ -0,0 +1,151 @@ +package rpc + +import ( + "testing" + + "github.com/gotd/td/tg" +) + +// TestScheduledGroupCallLifecycle 覆盖定时通话闭环：create(schedule_date) → +// scheduled 服务消息 + groupCall.schedule_date → 订阅开播提醒（per-viewer flag）→ +// startScheduledGroupCall 清 schedule_date + started 服务消息 → join 正常入会。 +func TestScheduledGroupCallLifecycle(t *testing.T) { + f := newGroupCallFixture(t) + ownerCtx := f.userCtx(f.owner, 11) + memberCtx := f.userCtx(f.member, 22) + scheduleDate := int(f.clk.Now().Unix()) + 3600 + + // --- 过去时间拒绝 --- + pastReq := &tg.PhoneCreateGroupCallRequest{ + Peer: &tg.InputPeerChannel{ChannelID: f.channel.ID, AccessHash: f.channel.AccessHash}, + RandomID: 1, + } + pastReq.SetScheduleDate(int(f.clk.Now().Unix()) - 10) + if _, err := f.router.onPhoneCreateGroupCall(ownerCtx, pastReq); err == nil { + t.Fatalf("past schedule_date accepted") + } else { + assertPhoneRPCErr(t, err, "SCHEDULE_DATE_INVALID") + } + + // --- create scheduled --- + createReq := &tg.PhoneCreateGroupCallRequest{ + Peer: &tg.InputPeerChannel{ChannelID: f.channel.ID, AccessHash: f.channel.AccessHash}, + RandomID: 2, + } + createReq.SetScheduleDate(scheduleDate) + createRes, err := f.router.onPhoneCreateGroupCall(ownerCtx, createReq) + if err != nil { + t.Fatalf("create scheduled group call: %v", err) + } + call := findUpdate[*tg.UpdateGroupCall](t, createRes).Call.(*tg.GroupCall) + if got, ok := call.GetScheduleDate(); !ok || got != scheduleDate { + t.Fatalf("groupCall.schedule_date = %d ok=%v, want %d", got, ok, scheduleDate) + } + // scheduled 服务消息。 + msgUpdate := findUpdate[*tg.UpdateNewChannelMessage](t, createRes) + svc, ok := msgUpdate.Message.(*tg.MessageService) + if !ok { + t.Fatalf("create response message = %T, want MessageService", msgUpdate.Message) + } + scheduledAction, ok := svc.Action.(*tg.MessageActionGroupCallScheduled) + if !ok { + t.Fatalf("service action = %T, want MessageActionGroupCallScheduled", svc.Action) + } + if scheduledAction.ScheduleDate != scheduleDate { + t.Fatalf("service action schedule_date = %d, want %d", scheduledAction.ScheduleDate, scheduleDate) + } + input := &tg.InputGroupCall{ID: call.ID, AccessHash: call.AccessHash} + + // --- member 订阅开播提醒 --- + subRes, err := f.router.onPhoneToggleGroupCallStartSubscription(memberCtx, &tg.PhoneToggleGroupCallStartSubscriptionRequest{ + Call: input, Subscribed: true, + }) + if err != nil { + t.Fatalf("toggle start subscription: %v", err) + } + subCall := findUpdate[*tg.UpdateGroupCall](t, subRes).Call.(*tg.GroupCall) + if !subCall.ScheduleStartSubscribed { + t.Fatalf("subscription response missing schedule_start_subscribed") + } + // getGroupCall 回填 per-viewer flag：member 已订阅、owner 未订阅。 + memberView, err := f.router.onPhoneGetGroupCall(memberCtx, &tg.PhoneGetGroupCallRequest{Call: input, Limit: 10}) + if err != nil { + t.Fatalf("member getGroupCall: %v", err) + } + if !memberView.Call.(*tg.GroupCall).ScheduleStartSubscribed { + t.Fatalf("member getGroupCall missing subscribed flag") + } + ownerView, err := f.router.onPhoneGetGroupCall(ownerCtx, &tg.PhoneGetGroupCallRequest{Call: input, Limit: 10}) + if err != nil { + t.Fatalf("owner getGroupCall: %v", err) + } + if ownerView.Call.(*tg.GroupCall).ScheduleStartSubscribed { + t.Fatalf("owner getGroupCall unexpectedly subscribed") + } + + // --- 非管理员不能开播 --- + if _, err := f.router.onPhoneStartScheduledGroupCall(memberCtx, input); err == nil { + t.Fatalf("non-admin started scheduled call") + } else { + assertPhoneRPCErr(t, err, "CHAT_ADMIN_REQUIRED") + } + + // --- start --- + f.sessions.reset() + startRes, err := f.router.onPhoneStartScheduledGroupCall(ownerCtx, input) + if err != nil { + t.Fatalf("startScheduledGroupCall: %v", err) + } + started := findUpdate[*tg.UpdateGroupCall](t, startRes).Call.(*tg.GroupCall) + if _, ok := started.GetScheduleDate(); ok { + t.Fatalf("started call still has schedule_date") + } + // started 服务消息（messageActionGroupCall 无 duration）。 + startMsg := findUpdate[*tg.UpdateNewChannelMessage](t, startRes) + startSvc := startMsg.Message.(*tg.MessageService) + if _, ok := startSvc.Action.(*tg.MessageActionGroupCall); !ok { + t.Fatalf("start service action = %T, want MessageActionGroupCall", startSvc.Action) + } + // 在线成员收到 schedule_date 已清的 updateGroupCall（客户端据此自动入会）。 + memberGotStart := false + for _, rec := range f.sessions.records() { + if rec.userID != f.member.ID { + continue + } + if box, ok := rec.msg.(*tg.Updates); ok { + for _, u := range box.Updates { + if gc, ok := u.(*tg.UpdateGroupCall); ok { + if call, ok := gc.Call.(*tg.GroupCall); ok { + if _, has := call.GetScheduleDate(); !has { + memberGotStart = true + } + } + } + } + } + } + if !memberGotStart { + t.Fatalf("member did not receive started updateGroupCall: %+v", f.sessions.records()) + } + + // --- 幂等重复 start --- + if _, err := f.router.onPhoneStartScheduledGroupCall(ownerCtx, input); err != nil { + t.Fatalf("idempotent re-start: %v", err) + } + + // --- 已开始后订阅提醒非法 --- + if _, err := f.router.onPhoneToggleGroupCallStartSubscription(memberCtx, &tg.PhoneToggleGroupCallStartSubscriptionRequest{ + Call: input, Subscribed: true, + }); err == nil { + t.Fatalf("subscription toggle allowed after start") + } + + // --- start 后正常 join --- + if _, err := f.router.onPhoneJoinGroupCall(memberCtx, &tg.PhoneJoinGroupCallRequest{ + Call: input, + JoinAs: &tg.InputPeerSelf{}, + Params: groupCallJoinParams(t, 7001), + }); err != nil { + t.Fatalf("join after start: %v", err) + } +} diff --git a/internal/rpc/phone_group_call_updates.go b/internal/rpc/phone_group_call_updates.go index 02144fdd..3d1b263f 100644 --- a/internal/rpc/phone_group_call_updates.go +++ b/internal/rpc/phone_group_call_updates.go @@ -53,14 +53,29 @@ func (r *Router) groupCallUpdateContainer(ctx context.Context, viewerUserID int6 } // pushGroupCallUpdate 把 updateGroupCall（call 行变化）推给在线群成员。 +// 定时通话需 per-viewer 回填 schedule_start_subscribed：TDesktop applyCallFields +// 无条件覆盖本地该 flag，漏填会把订阅者的"开播提醒"开关静默关掉。 func (r *Router) pushGroupCallUpdate(ctx context.Context, channel domain.Channel, call domain.GroupCall) { if call.Conference() { r.pushConferenceGroupCallUpdate(ctx, call) return } + var subscribed map[int64]struct{} + if call.ScheduleDate > 0 { + if ids, err := r.deps.GroupCalls.ScheduleSubscriberIDs(ctx, call.ID); err == nil { + subscribed = make(map[int64]struct{}, len(ids)) + for _, id := range ids { + subscribed[id] = struct{}{} + } + } + } recipients := r.groupCallOnlineRecipients(ctx, channel.ID) for _, viewerID := range recipients { - update := &tg.UpdateGroupCall{Call: tgGroupCall(call, viewerID, false)} + viewerCall := call + if subscribed != nil { + _, viewerCall.ScheduleStartSubscribed = subscribed[viewerID] + } + update := &tg.UpdateGroupCall{Call: tgGroupCall(viewerCall, viewerID, false)} update.SetPeer(&tg.PeerChannel{ChannelID: channel.ID}) r.pushUserMessage(ctx, viewerID, "group call update", r.groupCallUpdateContainer(ctx, viewerID, channel, update, []int64{call.CreatorUserID})) diff --git a/internal/rpc/phone_register.go b/internal/rpc/phone_register.go index 49dc5a56..0ae1125b 100644 --- a/internal/rpc/phone_register.go +++ b/internal/rpc/phone_register.go @@ -41,6 +41,9 @@ func (r *Router) registerPhone(d *tg.ServerDispatcher) { d.OnPhoneEditGroupCallTitle(r.onPhoneEditGroupCallTitle) d.OnPhoneToggleGroupCallSettings(r.onPhoneToggleGroupCallSettings) d.OnPhoneInviteToGroupCall(r.onPhoneInviteToGroupCall) + // 定时通话（scheduled video chat）。 + d.OnPhoneStartScheduledGroupCall(r.onPhoneStartScheduledGroupCall) + d.OnPhoneToggleGroupCallStartSubscription(r.onPhoneToggleGroupCallStartSubscription) // Ad-hoc E2E conference call（P2P 通话升级/拉人路径）。 d.OnPhoneCreateConferenceCall(r.onPhoneCreateConferenceCall) d.OnPhoneInviteConferenceCallParticipant(r.onPhoneInviteConferenceCallParticipant) diff --git a/internal/rpc/phone_stubs.go b/internal/rpc/phone_stubs.go index 34845be7..0b6c25e4 100644 --- a/internal/rpc/phone_stubs.go +++ b/internal/rpc/phone_stubs.go @@ -10,18 +10,9 @@ import ( // 通话内消息族 / scheduled / RTMP）走 router fallback：400/500 NOT_IMPLEMENTED + // 兼容矩阵日志，客户端不断连。 func (r *Router) registerPhoneStubs(d *tg.ServerDispatcher) { - // 入会面板前置调用：返回 self 一个候选身份（空返回会卡 UI）。 - d.OnPhoneGetGroupCallJoinAs(func(ctx context.Context, peer tg.InputPeerClass) (*tg.PhoneJoinAsPeers, error) { - userID, err := r.phoneRequireUser(ctx) - if err != nil { - return nil, err - } - return &tg.PhoneJoinAsPeers{ - Peers: []tg.PeerClass{&tg.PeerUser{UserID: userID}}, - Chats: []tg.ChatClass{}, - Users: r.tgUsersForIDs(ctx, userID, []int64{userID}), - }, nil - }) + // 入会身份候选：真实实现见 phone_group_call.go（self + admin 的频道身份）。 + d.OnPhoneGetGroupCallJoinAs(r.onPhoneGetGroupCallJoinAs) + // default join-as 偏好持久化仍是 stub（chatFull.groupcall_default_join_as 不回填）。 d.OnPhoneSaveDefaultGroupCallJoinAs(func(ctx context.Context, req *tg.PhoneSaveDefaultGroupCallJoinAsRequest) (bool, error) { return true, nil }) @@ -29,8 +20,7 @@ func (r *Router) registerPhoneStubs(d *tg.ServerDispatcher) { d.OnPhoneToggleGroupCallRecord(func(ctx context.Context, req *tg.PhoneToggleGroupCallRecordRequest) (tg.UpdatesClass, error) { return tgEmptyUpdates(int(r.clock.Now().Unix())), nil }) - // RTMP 直播范围外。 - d.OnPhoneGetGroupCallStreamChannels(func(ctx context.Context, call tg.InputGroupCallClass) (*tg.PhoneGroupCallStreamChannels, error) { - return &tg.PhoneGroupCallStreamChannels{Channels: []tg.GroupCallStreamChannel{}}, nil - }) + // RTMP 直播（Live Stream）：真实 handler 见 phone_group_call_rtmp.go。 + d.OnPhoneGetGroupCallStreamChannels(r.onPhoneGetGroupCallStreamChannels) + d.OnPhoneGetGroupCallStreamRtmpURL(r.onPhoneGetGroupCallStreamRtmpURL) } diff --git a/internal/rpc/router.go b/internal/rpc/router.go index 210e7f05..d2e6d918 100644 --- a/internal/rpc/router.go +++ b/internal/rpc/router.go @@ -67,6 +67,9 @@ type Config struct { CallForceRelay bool // GroupCallMaxParticipants 是群通话单房间参与者上限；<=0 不限制。 GroupCallMaxParticipants int + // RtmpIngestURL 是 getGroupCallStreamRtmpUrl 返回给推流端（OBS）的服务器地址， + // 形如 "rtmp://:/live"。为空时回落 "rtmp://:2400/live"。 + RtmpIngestURL string // TempKeyResolveCacheTTL 是 PFS temp→perm auth key 解析的进程内缓存有效期。>0 时同一 temp key // 在 TTL 内复用上次解析、跳过每帧 ResolveAuthKey 的 PG 查询；0（默认/测试）关闭=每帧重校验。 // 显式撤销会删除协议 auth key、清缓存并断开活跃连接；TTL 只影响自然过期或异常路径下的 @@ -163,6 +166,9 @@ func New(cfg Config, deps Deps, log *zap.Logger, clk clock.Clock) *Router { r := &Router{cfg: cfg, log: log, clock: clk, deps: deps, presence: newPresenceTracker(), callbacks: newCallbackRegistry(), inlines: newInlineRegistry(botInlineQueryTTL, deps.Inline), webviews: newWebViewRegistry(webViewSessionTTL, deps.Inline), loginTokens: newLoginTokenRegistry(), tempKeyResolveCache: newTempKeyResolveCache(cfg.TempKeyResolveCacheMaxEntries), storyProjectionCache: newStoryProjectionCache(clk.Now), storyPinnedCache: newStoryPinnedAvailableCache(clk.Now), storyPinnedListCache: newStoryPinnedStoriesCache(clk.Now), channelFullBotCache: newChannelFullBotInfoCache(clk.Now), userFullProjectionCache: newUserFullProjectionCache(clk.Now), peerSettingsProjectionCache: newPeerSettingsProjectionCache(clk.Now), channelFullProjectionCache: newChannelFullProjectionCache(clk.Now), emojiStickers: newEmojiStickerIndex(clk.Now), notifySettings: newNotifySettingsCache(clk.Now), stickerCatalog: newStickerCatalogCache(clk.Now), accountSettings: newAccountSettingsCache(clk.Now), instanceID: instanceID} r.channelFanout = newChannelFanoutDispatcher(r, defaultChannelFanoutShards, defaultChannelFanoutBuffer) r.webPageResolveSem = make(chan struct{}, webPageResolveConcurrency) + if cfg.DC > 0 { + groupCallStreamDCID = cfg.DC + } d := tg.NewServerDispatcher(r.fallback) r.registerHelp(d) diff --git a/internal/rpc/sticker_link_smoke_test.go b/internal/rpc/sticker_link_smoke_test.go index 1ef69c26..b6b9a5fd 100644 --- a/internal/rpc/sticker_link_smoke_test.go +++ b/internal/rpc/sticker_link_smoke_test.go @@ -201,7 +201,7 @@ func TestStickersBotCreatePackLinkInstallIsolationSmoke(t *testing.T) { sendStickersBotText(t, r, alice, "Alice Bot Pack", 9102) waitForStickersReply(t, messageStore, alice.ID, "Lottie JSON") sendStickersBotDocument(t, r, alice, 401, 4401, 9103) - waitForStickersReply(t, messageStore, alice.ID, "emoji") + waitForStickersReply(t, messageStore, alice.ID, "Now send the emoji") sendStickersBotText(t, r, alice, "🙂", 9104) waitForStickersReply(t, messageStore, alice.ID, "Added") sendStickersBotText(t, r, alice, "/publish", 9105) diff --git a/internal/rpc/upload.go b/internal/rpc/upload.go index 0a4cfaad..4185072b 100644 --- a/internal/rpc/upload.go +++ b/internal/rpc/upload.go @@ -7,6 +7,7 @@ import ( "strings" "github.com/gotd/td/tg" + "go.uber.org/zap" "telesrv/internal/domain" ) @@ -67,12 +68,17 @@ func (r *Router) onUploadSaveBigFilePart(ctx context.Context, req *tg.UploadSave } func (r *Router) onUploadGetFile(ctx context.Context, req *tg.UploadGetFileRequest) (tg.UploadFileClass, error) { - if r.deps.Files == nil { - return nil, notImplementedErr() - } if req.Offset < 0 || req.Limit <= 0 || req.Limit > maxUploadGetFileChunkLimit { return nil, limitInvalidErr() } + // RTMP 直播拉流：inputGroupCallStream 不落 file_blobs、不依赖 Files 服务， + // 直连 livestream 媒体面（须先于 Files nil 检查）。 + if loc, ok := req.Location.(*tg.InputGroupCallStream); ok { + return r.onUploadGetGroupCallStream(ctx, loc, req.Offset, req.Limit) + } + if r.deps.Files == nil { + return nil, notImplementedErr() + } key, ok := fileLocationKey(req.Location) if !ok { return nil, locationInvalidErr() @@ -95,6 +101,54 @@ func (r *Router) onUploadGetFile(ctx context.Context, req *tg.UploadGetFileReque return nil, locationInvalidErr() } +// onUploadGetGroupCallStream 处理 RTMP 直播观众拉流：按 time_ms/scale 取一段打包好的 +// tgcalls broadcast part，再按 offset/limit 切片返回。错误语义对齐 TDesktop 消费点 +// （calls_group_call.cpp broadcastPartStart）： +// - 段未就绪（时间轴还没走到）→ TIME_TOO_BIG（客户端 100ms 后原样重试） +// - 段已过期/无流/未加入 → GROUPCALL_JOIN_MISSING（触发客户端 rejoin 重新对时） +func (r *Router) onUploadGetGroupCallStream(ctx context.Context, loc *tg.InputGroupCallStream, offset int64, limit int) (tg.UploadFileClass, error) { + if r.deps.LiveStreams == nil { + return nil, notImplementedErr() + } + scope, err := r.groupCallScopeFrom(ctx, loc.Call) + if err != nil { + return nil, err + } + if !scope.call.RtmpStream || scope.channel.ID == 0 { + return nil, groupCallInvalidErr() + } + // RTMP 观众在 stream 模式不发 checkGroupCall 心跳、也无 SFU 媒体面活性， + // 拉流请求（~1/s/观众）就是它的保活信号——不刷会被 sweeper 置 left， + // 客户端每 ~50s 报 "Rejoin after got 'left' with my ssrc" 循环重进。 + if _, _, err := r.deps.GroupCalls.Touch(ctx, scope.call.ID, scope.userID, int(r.clock.Now().Unix())); err != nil { + r.log.Debug("live stream viewer touch", zap.Int64("call_id", scope.call.ID), zap.Error(err)) + } + part, err := r.deps.LiveStreams.StreamPart(scope.channel.ID, loc.TimeMs, loc.Scale) + switch { + case errors.Is(err, domain.ErrLiveStreamPartNotReady): + // 时间轴尚未走到该段：客户端 100ms 后原样重试（Status::NotReady）。 + return nil, tgerr400("TIME_TOO_BIG") + case errors.Is(err, domain.ErrLiveStreamPartExpired), errors.Is(err, domain.ErrLiveStreamNoStream): + // 段已淘汰/无流：客户端重新对时后 resync（Status::ResyncNeeded）。 + return nil, tgerr400("STREAM_TIMESTAMP_EXPIRED") + case err != nil: + return nil, internalErr() + } + // offset 越界返回空 bytes（客户端已读完该段即停止续读，limit=128KiB 单次到底）。 + if offset >= int64(len(part)) { + return &tg.UploadFile{Type: &tg.StorageFileUnknown{}, Bytes: []byte{}}, nil + } + end := offset + int64(limit) + if end > int64(len(part)) { + end = int64(len(part)) + } + return &tg.UploadFile{ + Type: &tg.StorageFileUnknown{}, + Mtime: 0, + Bytes: part[offset:end], + }, nil +} + // onUploadGetFileHashes 返回空 hash 列表：本阶段不做 CDN/分片完整性校验，客户端据空列表直接信任数据。 func (r *Router) onUploadGetFileHashes(ctx context.Context, req *tg.UploadGetFileHashesRequest) ([]tg.FileHash, error) { return []tg.FileHash{}, nil diff --git a/internal/store/groupcall.go b/internal/store/groupcall.go index 9dc312d2..2c11a5aa 100644 --- a/internal/store/groupcall.go +++ b/internal/store/groupcall.go @@ -63,4 +63,15 @@ type GroupCallStore interface { ListConferenceRecipientUserIDs(ctx context.Context, callID int64) ([]int64, error) AppendGroupCallChainBlock(ctx context.Context, block domain.GroupCallChainBlock) (domain.GroupCallChainBlock, error) ListGroupCallChainBlocks(ctx context.Context, callID int64, subChainID, offset, limit int) (domain.GroupCallChainBlockPage, error) + // GetRtmpStreamKey / SetRtmpStreamKey 维护 per-channel 的持久 RTMP 推流密钥。 + // revoke 语义由上层实现：Set 覆盖旧 key，旧 key 推流即刻失效。 + GetRtmpStreamKey(ctx context.Context, channelID int64) (string, bool, error) + SetRtmpStreamKey(ctx context.Context, channelID int64, key string, now int) error + // StartScheduledGroupCall 清零 schedule_date（定时通话正式开始）。 + // changed=false 表示本来就已开始（幂等）；discarded 返回 ErrGroupCallDiscarded。 + StartScheduledGroupCall(ctx context.Context, callID int64) (domain.GroupCall, bool, error) + // SetScheduleStartSubscription 写入/清除 userID 的开播提醒订阅。 + SetScheduleStartSubscription(ctx context.Context, callID, userID int64, subscribed bool) error + // ListScheduleSubscriberIDs 返回订阅了开播提醒的 userID（升序）。 + ListScheduleSubscriberIDs(ctx context.Context, callID int64) ([]int64, error) } diff --git a/internal/store/memory/groupcall.go b/internal/store/memory/groupcall.go index 1c076818..abc664d4 100644 --- a/internal/store/memory/groupcall.go +++ b/internal/store/memory/groupcall.go @@ -45,7 +45,9 @@ type GroupCallStore struct { inviteByMessage map[inviteMessageKey]domain.GroupCallInvite chainBlocks map[chainKey][]domain.GroupCallChainBlock overrides map[overrideKey]domain.GroupCallParticipantOverride - raiseHandSeq map[int64]int64 // callID → 单调举手序号 + raiseHandSeq map[int64]int64 // callID → 单调举手序号 + rtmpKeys map[int64]string // channelID → RTMP 推流密钥 + scheduleSubs map[int64]map[int64]struct{} // callID → 订阅开播提醒的 userID nextSyntheticID int64 } @@ -62,6 +64,8 @@ func NewGroupCallStore() *GroupCallStore { chainBlocks: make(map[chainKey][]domain.GroupCallChainBlock), overrides: make(map[overrideKey]domain.GroupCallParticipantOverride), raiseHandSeq: make(map[int64]int64), + rtmpKeys: make(map[int64]string), + scheduleSubs: make(map[int64]map[int64]struct{}), } } @@ -176,12 +180,13 @@ func (s *GroupCallStore) JoinGroupCall(_ context.Context, req domain.JoinGroupCa existing, rejoining := rows[req.UserID] wasActive := rejoining && !existing.Left p := domain.GroupCallParticipant{ - CallID: req.CallID, - UserID: req.UserID, - SSRC: req.SSRC, - JoinDate: req.Now, - ActiveDate: req.Now, - LastCheckDate: req.Now, + CallID: req.CallID, + UserID: req.UserID, + JoinAsChannelID: req.JoinAsChannelID, + SSRC: req.SSRC, + JoinDate: req.Now, + ActiveDate: req.Now, + LastCheckDate: req.Now, // VideoJSON 整体替换、PresentationJSON 随全新行清空（rejoin 后客户端 // 会重发 joinGroupCallPresentation，旧屏幕登记必须作废）。 VideoJSON: append([]byte(nil), req.VideoJSON...), @@ -803,6 +808,72 @@ func (s *GroupCallStore) ListGroupCallChainBlocks(_ context.Context, callID int6 return page, nil } +func (s *GroupCallStore) StartScheduledGroupCall(_ context.Context, callID int64) (domain.GroupCall, bool, error) { + s.mu.Lock() + defer s.mu.Unlock() + call, ok := s.calls[callID] + if !ok { + return domain.GroupCall{}, false, domain.ErrGroupCallInvalid + } + if !call.Active() { + return domain.GroupCall{}, false, domain.ErrGroupCallDiscarded + } + if call.ScheduleDate == 0 { + return call, false, nil + } + call.ScheduleDate = 0 + s.calls[callID] = call + return call, true, nil +} + +func (s *GroupCallStore) SetScheduleStartSubscription(_ context.Context, callID, userID int64, subscribed bool) error { + if callID == 0 || userID == 0 { + return domain.ErrGroupCallInvalid + } + s.mu.Lock() + defer s.mu.Unlock() + subs := s.scheduleSubs[callID] + if subscribed { + if subs == nil { + subs = make(map[int64]struct{}) + s.scheduleSubs[callID] = subs + } + subs[userID] = struct{}{} + return nil + } + delete(subs, userID) + return nil +} + +func (s *GroupCallStore) ListScheduleSubscriberIDs(_ context.Context, callID int64) ([]int64, error) { + s.mu.Lock() + defer s.mu.Unlock() + subs := s.scheduleSubs[callID] + out := make([]int64, 0, len(subs)) + for id := range subs { + out = append(out, id) + } + sort.Slice(out, func(i, j int) bool { return out[i] < out[j] }) + return out, nil +} + +func (s *GroupCallStore) GetRtmpStreamKey(_ context.Context, channelID int64) (string, bool, error) { + s.mu.Lock() + defer s.mu.Unlock() + key, ok := s.rtmpKeys[channelID] + return key, ok, nil +} + +func (s *GroupCallStore) SetRtmpStreamKey(_ context.Context, channelID int64, key string, _ int) error { + if channelID == 0 || key == "" { + return domain.ErrGroupCallInvalid + } + s.mu.Lock() + defer s.mu.Unlock() + s.rtmpKeys[channelID] = key + return nil +} + func max(a, b int) int { if a > b { return a diff --git a/internal/store/postgres/groupcall.go b/internal/store/postgres/groupcall.go index 33719be0..5c064109 100644 --- a/internal/store/postgres/groupcall.go +++ b/internal/store/postgres/groupcall.go @@ -27,9 +27,9 @@ func NewGroupCallStore(db sqlcgen.DBTX) *GroupCallStore { const groupCallColumns = `call_id, access_hash, channel_id, creator_user_id, kind, state, title, join_muted, version, participants_count, created_at, discarded_at, duration, started_msg_id, -invite_slug, invite_link, random_id, migrated_from_phone_call_id` +invite_slug, invite_link, random_id, migrated_from_phone_call_id, rtmp_stream, schedule_date` -const groupCallParticipantColumns = `call_id, user_id, ssrc, join_date, active_date, muted, muted_by_admin, +const groupCallParticipantColumns = `call_id, user_id, join_as_channel_id, ssrc, join_date, active_date, muted, muted_by_admin, volume_by_admin, raise_hand_rating, video_json, presentation_json, public_key, join_block, left_call, last_check_date` func scanGroupCall(row rowScanner) (domain.GroupCall, error) { @@ -38,7 +38,7 @@ func scanGroupCall(row rowScanner) (domain.GroupCall, error) { if err := row.Scan( &c.ID, &c.AccessHash, &c.ChannelID, &c.CreatorUserID, &kind, &state, &c.Title, &c.JoinMuted, &c.Version, &c.ParticipantsCount, &c.CreatedAt, &c.DiscardedAt, &c.Duration, &c.StartedMsgID, - &c.InviteSlug, &c.InviteLink, &c.RandomID, &c.MigratedFromPhoneCallID, + &c.InviteSlug, &c.InviteLink, &c.RandomID, &c.MigratedFromPhoneCallID, &c.RtmpStream, &c.ScheduleDate, ); err != nil { return domain.GroupCall{}, err } @@ -50,7 +50,7 @@ func scanGroupCall(row rowScanner) (domain.GroupCall, error) { func scanGroupCallParticipant(row rowScanner) (domain.GroupCallParticipant, error) { var p domain.GroupCallParticipant if err := row.Scan( - &p.CallID, &p.UserID, &p.SSRC, &p.JoinDate, &p.ActiveDate, &p.Muted, &p.MutedByAdmin, + &p.CallID, &p.UserID, &p.JoinAsChannelID, &p.SSRC, &p.JoinDate, &p.ActiveDate, &p.Muted, &p.MutedByAdmin, &p.VolumeByAdmin, &p.RaiseHandRating, &p.VideoJSON, &p.PresentationJSON, &p.PublicKey, &p.JoinBlock, &p.Left, &p.LastCheckDate, ); err != nil { return domain.GroupCallParticipant{}, err @@ -73,7 +73,7 @@ func scanGroupCallInviteJoined(row rowScanner) (domain.GroupCall, domain.GroupCa if err := row.Scan( &c.ID, &c.AccessHash, &c.ChannelID, &c.CreatorUserID, &kind, &state, &c.Title, &c.JoinMuted, &c.Version, &c.ParticipantsCount, &c.CreatedAt, &c.DiscardedAt, &c.Duration, &c.StartedMsgID, - &c.InviteSlug, &c.InviteLink, &c.RandomID, &c.MigratedFromPhoneCallID, + &c.InviteSlug, &c.InviteLink, &c.RandomID, &c.MigratedFromPhoneCallID, &c.RtmpStream, &c.ScheduleDate, &inv.CallID, &inv.InviterUserID, &inv.InviteeUserID, &inv.MessageID, &status, &inv.Video, &inv.CreatedAt, &inv.UpdatedAt, ); err != nil { return domain.GroupCall{}, domain.GroupCallInvite{}, err @@ -155,9 +155,9 @@ func (s *GroupCallStore) CreateGroupCall(ctx context.Context, call domain.GroupC call.Version = 1 } _, err := s.db.Exec(ctx, ` -INSERT INTO group_calls (call_id, access_hash, channel_id, creator_user_id, kind, state, title, join_muted, version, participants_count, created_at) -VALUES ($1, $2, $3, $4, 'channel', 'active', $5, $6, $7, 0, $8)`, - call.ID, call.AccessHash, call.ChannelID, call.CreatorUserID, call.Title, call.JoinMuted, call.Version, call.CreatedAt) +INSERT INTO group_calls (call_id, access_hash, channel_id, creator_user_id, kind, state, title, join_muted, version, participants_count, created_at, rtmp_stream, schedule_date) +VALUES ($1, $2, $3, $4, 'channel', 'active', $5, $6, $7, 0, $8, $9, $10)`, + call.ID, call.AccessHash, call.ChannelID, call.CreatorUserID, call.Title, call.JoinMuted, call.Version, call.CreatedAt, call.RtmpStream, call.ScheduleDate) if err != nil { var pgErr *pgconn.PgError if errors.As(err, &pgErr) && pgErr.Code == "23505" { @@ -299,14 +299,15 @@ func (s *GroupCallStore) JoinGroupCall(ctx context.Context, req domain.JoinGroup return domain.GroupCallMutation{}, fmt.Errorf("load group call participant: %w", err) } p := domain.GroupCallParticipant{ - CallID: req.CallID, - UserID: req.UserID, - SSRC: req.SSRC, - JoinDate: req.Now, - ActiveDate: req.Now, - LastCheckDate: req.Now, - PublicKey: append([]byte(nil), req.PublicKey...), - JoinBlock: append([]byte(nil), req.JoinBlock...), + CallID: req.CallID, + UserID: req.UserID, + JoinAsChannelID: req.JoinAsChannelID, + SSRC: req.SSRC, + JoinDate: req.Now, + ActiveDate: req.Now, + LastCheckDate: req.Now, + PublicKey: append([]byte(nil), req.PublicKey...), + JoinBlock: append([]byte(nil), req.JoinBlock...), } if wasActive { // 同人换 ssrc 的 rejoin 保留原 join_date（列表排序稳定）。 @@ -320,9 +321,10 @@ func (s *GroupCallStore) JoinGroupCall(ctx context.Context, req domain.JoinGroup // video_json 整体替换、presentation_json 清空（rejoin 后客户端会重发 // joinGroupCallPresentation，旧屏幕登记必须作废）。 if _, err := tx.Exec(ctx, ` -INSERT INTO group_call_participants (call_id, user_id, ssrc, join_date, active_date, muted, muted_by_admin, volume_by_admin, raise_hand_rating, video_json, public_key, join_block, left_call, last_check_date) -VALUES ($1, $2, $3, $4, $5, $6, $7, 0, 0, $8, $9, $10, FALSE, $11) +INSERT INTO group_call_participants (call_id, user_id, join_as_channel_id, ssrc, join_date, active_date, muted, muted_by_admin, volume_by_admin, raise_hand_rating, video_json, public_key, join_block, left_call, last_check_date) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, 0, 0, $9, $10, $11, FALSE, $12) ON CONFLICT (call_id, user_id) DO UPDATE SET + join_as_channel_id = EXCLUDED.join_as_channel_id, ssrc = EXCLUDED.ssrc, join_date = EXCLUDED.join_date, active_date = EXCLUDED.active_date, @@ -336,7 +338,7 @@ ON CONFLICT (call_id, user_id) DO UPDATE SET join_block = EXCLUDED.join_block, left_call = FALSE, last_check_date = EXCLUDED.last_check_date`, - req.CallID, req.UserID, req.SSRC, p.JoinDate, p.ActiveDate, p.Muted, p.MutedByAdmin, + req.CallID, req.UserID, req.JoinAsChannelID, req.SSRC, p.JoinDate, p.ActiveDate, p.Muted, p.MutedByAdmin, nullableJSON(p.VideoJSON), nullableGroupCallBytes(p.PublicKey), nullableGroupCallBytes(p.JoinBlock), p.LastCheckDate); err != nil { var pgErr *pgconn.PgError if errors.As(err, &pgErr) && pgErr.Code == "23505" { @@ -1147,6 +1149,108 @@ RETURNING call_id, sub_chain_id, block_offset, author_user_id, block, created_at return block, nil } +func (s *GroupCallStore) GetRtmpStreamKey(ctx context.Context, channelID int64) (string, bool, error) { + var key string + err := s.db.QueryRow(ctx, + `SELECT stream_key FROM group_call_rtmp_keys WHERE channel_id = $1`, channelID).Scan(&key) + if errors.Is(err, pgx.ErrNoRows) { + return "", false, nil + } + if err != nil { + return "", false, fmt.Errorf("get rtmp stream key: %w", err) + } + return key, true, nil +} + +func (s *GroupCallStore) SetRtmpStreamKey(ctx context.Context, channelID int64, key string, now int) error { + if channelID == 0 || key == "" { + return domain.ErrGroupCallInvalid + } + if _, err := s.db.Exec(ctx, ` +INSERT INTO group_call_rtmp_keys (channel_id, stream_key, updated_at) +VALUES ($1, $2, $3) +ON CONFLICT (channel_id) DO UPDATE SET stream_key = EXCLUDED.stream_key, updated_at = EXCLUDED.updated_at`, + channelID, key, now); err != nil { + return fmt.Errorf("set rtmp stream key: %w", err) + } + return nil +} + +func (s *GroupCallStore) StartScheduledGroupCall(ctx context.Context, callID int64) (domain.GroupCall, bool, error) { + tx, err := s.begin(ctx, "start scheduled group call") + if err != nil { + return domain.GroupCall{}, false, err + } + committed := false + defer func() { + if !committed { + _ = tx.Rollback(ctx) + } + }() + call, err := lockGroupCallTx(ctx, tx, callID) + if err != nil { + return domain.GroupCall{}, false, err + } + if !call.Active() { + return domain.GroupCall{}, false, domain.ErrGroupCallDiscarded + } + if call.ScheduleDate == 0 { + // 幂等：已开始。 + if err := tx.Commit(ctx); err != nil { + return domain.GroupCall{}, false, fmt.Errorf("commit start scheduled noop: %w", err) + } + committed = true + return call, false, nil + } + call, err = scanGroupCall(tx.QueryRow(ctx, ` +UPDATE group_calls SET schedule_date = 0 WHERE call_id = $1 RETURNING `+groupCallColumns, callID)) + if err != nil { + return domain.GroupCall{}, false, fmt.Errorf("start scheduled group call: %w", err) + } + if err := tx.Commit(ctx); err != nil { + return domain.GroupCall{}, false, fmt.Errorf("commit start scheduled group call: %w", err) + } + committed = true + return call, true, nil +} + +func (s *GroupCallStore) SetScheduleStartSubscription(ctx context.Context, callID, userID int64, subscribed bool) error { + if callID == 0 || userID == 0 { + return domain.ErrGroupCallInvalid + } + if !subscribed { + if _, err := s.db.Exec(ctx, + `DELETE FROM group_call_schedule_subscribers WHERE call_id = $1 AND user_id = $2`, callID, userID); err != nil { + return fmt.Errorf("clear schedule subscription: %w", err) + } + return nil + } + if _, err := s.db.Exec(ctx, ` +INSERT INTO group_call_schedule_subscribers (call_id, user_id) +VALUES ($1, $2) ON CONFLICT DO NOTHING`, callID, userID); err != nil { + return fmt.Errorf("set schedule subscription: %w", err) + } + return nil +} + +func (s *GroupCallStore) ListScheduleSubscriberIDs(ctx context.Context, callID int64) ([]int64, error) { + rows, err := s.db.Query(ctx, + `SELECT user_id FROM group_call_schedule_subscribers WHERE call_id = $1 ORDER BY user_id`, callID) + if err != nil { + return nil, fmt.Errorf("list schedule subscribers: %w", err) + } + defer rows.Close() + var out []int64 + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + return nil, err + } + out = append(out, id) + } + return out, rows.Err() +} + func (s *GroupCallStore) ListGroupCallChainBlocks(ctx context.Context, callID int64, subChainID, offset, limit int) (domain.GroupCallChainBlockPage, error) { if limit <= 0 || limit > 100 { limit = 100