keyboard/keyBoard/Class/AiTalk/VM/TTSPlaybackPipeline.m

//
//  TTSPlaybackPipeline.m
//  keyBoard
//
//  Created by Mac on 2026/1/15.
//

#import "TTSPlaybackPipeline.h"
#import "AudioStreamPlayer.h"
#import <AVFoundation/AVFoundation.h>

@interface TTSPlaybackPipeline () <AudioStreamPlayerDelegate>

// 播放器
@property(nonatomic, strong) AVPlayer *urlPlayer;
@property(nonatomic, strong) AudioStreamPlayer *streamPlayer;

// 片段队列
@property(nonatomic, strong) NSMutableArray<NSDictionary *> *segmentQueue;
@property(nonatomic, strong)
    NSMutableDictionary<NSString *, NSNumber *> *segmentDurations;

// 状态
@property(nonatomic, assign) BOOL playing;
@property(nonatomic, copy) NSString *currentSegmentId;
@property(nonatomic, strong) id playerTimeObserver;

// 队列
@property(nonatomic, strong) dispatch_queue_t playbackQueue;

@end

@implementation TTSPlaybackPipeline

- (instancetype)init {
  self = [super init];
  if (self) {
    _segmentQueue = [[NSMutableArray alloc] init];
    _segmentDurations = [[NSMutableDictionary alloc] init];
    _playbackQueue = dispatch_queue_create("com.keyboard.aitalk.playback",
                                           DISPATCH_QUEUE_SERIAL);
  }
  return self;
}

- (void)dealloc {
  [self stop];
}

#pragma mark - Public Methods

- (BOOL)start:(NSError **)error {
  // 初始化 stream player
  if (!self.streamPlayer) {
    self.streamPlayer = [[AudioStreamPlayer alloc] init];
    self.streamPlayer.delegate = self;
  }

  return [self.streamPlayer start:error];
}

- (void)stop {
  dispatch_async(self.playbackQueue, ^{
    // 停止 URL 播放
    if (self.urlPlayer) {
      [self.urlPlayer pause];
      if (self.playerTimeObserver) {
        [self.urlPlayer removeTimeObserver:self.playerTimeObserver];
        self.playerTimeObserver = nil;
      }
      self.urlPlayer = nil;
    }

    // 停止流式播放
    [self.streamPlayer stop];

    // 清空队列
    [self.segmentQueue removeAllObjects];
    [self.segmentDurations removeAllObjects];

    self.playing = NO;
    self.currentSegmentId = nil;
  });
}

- (void)enqueueURL:(NSURL *)url segmentId:(NSString *)segmentId {
  if (!url || !segmentId)
    return;

  dispatch_async(self.playbackQueue, ^{
    NSDictionary *segment = @{
      @"type" : @(TTSPayloadTypeURL),
      @"url" : url,
      @"segmentId" : segmentId
    };
    [self.segmentQueue addObject:segment];

    // 如果当前没有在播放，开始播放
    if (!self.playing) {
      [self playNextSegment];
    }
  });
}

- (void)enqueueChunk:(NSData *)chunk
         payloadType:(TTSPayloadType)type
           segmentId:(NSString *)segmentId {
  if (!chunk || !segmentId)
    return;

  dispatch_async(self.playbackQueue, ^{
    switch (type) {
    case TTSPayloadTypePCMChunk:
      // 直接喂给 stream player
      [self.streamPlayer enqueuePCMChunk:chunk
                              sampleRate:16000
                                channels:1
                               segmentId:segmentId];

      if (!self.playing) {
        self.playing = YES;
        self.currentSegmentId = segmentId;

        dispatch_async(dispatch_get_main_queue(), ^{
          if ([self.delegate respondsToSelector:@selector
                             (pipelineDidStartSegment:duration:)]) {
            [self.delegate pipelineDidStartSegment:segmentId duration:0];
          }
        });
      }
      break;

    case TTSPayloadTypeAACChunk:
      // TODO: AAC 解码 -> PCM -> streamPlayer
      NSLog(@"[TTSPlaybackPipeline] AAC chunk decoding not implemented yet");
      break;

    case TTSPayloadTypeOpusChunk:
      // TODO: Opus 解码 -> PCM -> streamPlayer
      NSLog(@"[TTSPlaybackPipeline] Opus chunk decoding not implemented yet");
      break;

    default:
      break;
    }
  });
}

- (void)markSegmentComplete:(NSString *)segmentId {
  // Stream player 会自动处理播放完成
}

- (NSTimeInterval)currentTimeForSegment:(NSString *)segmentId {
  if (![segmentId isEqualToString:self.currentSegmentId]) {
    return 0;
  }

  if (self.urlPlayer) {
    return CMTimeGetSeconds(self.urlPlayer.currentTime);
  }

  return [self.streamPlayer playbackTimeForSegment:segmentId];
}

- (NSTimeInterval)durationForSegment:(NSString *)segmentId {
  NSNumber *duration = self.segmentDurations[segmentId];
  if (duration) {
    return duration.doubleValue;
  }

  if (self.urlPlayer && [segmentId isEqualToString:self.currentSegmentId]) {
    CMTime duration = self.urlPlayer.currentItem.duration;
    if (CMTIME_IS_VALID(duration)) {
      return CMTimeGetSeconds(duration);
    }
  }

  return [self.streamPlayer durationForSegment:segmentId];
}

#pragma mark - Private Methods

- (void)playNextSegment {
  if (self.segmentQueue.count == 0) {
    self.playing = NO;
    self.currentSegmentId = nil;

    dispatch_async(dispatch_get_main_queue(), ^{
      if ([self.delegate
              respondsToSelector:@selector(pipelineDidFinishAllSegments)]) {
        [self.delegate pipelineDidFinishAllSegments];
      }
    });
    return;
  }

  NSDictionary *segment = self.segmentQueue.firstObject;
  [self.segmentQueue removeObjectAtIndex:0];

  TTSPayloadType type = [segment[@"type"] integerValue];
  NSString *segmentId = segment[@"segmentId"];

  self.playing = YES;
  self.currentSegmentId = segmentId;

  if (type == TTSPayloadTypeURL) {
    NSURL *url = segment[@"url"];
    [self playURL:url segmentId:segmentId];
  }
}

- (void)playURL:(NSURL *)url segmentId:(NSString *)segmentId {
  AVPlayerItem *item = [AVPlayerItem playerItemWithURL:url];

  if (!self.urlPlayer) {
    self.urlPlayer = [AVPlayer playerWithPlayerItem:item];
  } else {
    [self.urlPlayer replaceCurrentItemWithPlayerItem:item];
  }

  // 监听播放完成
  [[NSNotificationCenter defaultCenter]
      addObserver:self
         selector:@selector(playerItemDidFinish:)
             name:AVPlayerItemDidPlayToEndTimeNotification
           object:item];

  // 添加时间观察器
  __weak typeof(self) weakSelf = self;
  self.playerTimeObserver = [self.urlPlayer
      addPeriodicTimeObserverForInterval:CMTimeMake(1, 30)
                                   queue:dispatch_get_main_queue()
                              usingBlock:^(CMTime time) {
                                __strong typeof(weakSelf) strongSelf = weakSelf;
                                if (!strongSelf)
                                  return;

                                NSTimeInterval currentTime =
                                    CMTimeGetSeconds(time);
                                if ([strongSelf.delegate
                                        respondsToSelector:@selector
                                        (pipelineDidUpdatePlaybackTime:
                                                             segmentId:)]) {
                                  [strongSelf.delegate
                                      pipelineDidUpdatePlaybackTime:currentTime
                                                          segmentId:segmentId];
                                }
                              }];

  // 等待资源加载后获取时长并开始播放
  [item.asset
      loadValuesAsynchronouslyForKeys:@[ @"duration" ]
                    completionHandler:^{
                      dispatch_async(dispatch_get_main_queue(), ^{
                        NSTimeInterval duration =
                            CMTimeGetSeconds(item.duration);
                        if (!isnan(duration)) {
                          self.segmentDurations[segmentId] = @(duration);
                        }

                        if ([self.delegate respondsToSelector:@selector
                                           (pipelineDidStartSegment:
                                                           duration:)]) {
                          [self.delegate pipelineDidStartSegment:segmentId
                                                        duration:duration];
                        }

                        [self.urlPlayer play];
                      });
                    }];
}

- (void)playerItemDidFinish:(NSNotification *)notification {
  [[NSNotificationCenter defaultCenter]
      removeObserver:self
                name:AVPlayerItemDidPlayToEndTimeNotification
              object:notification.object];

  if (self.playerTimeObserver) {
    [self.urlPlayer removeTimeObserver:self.playerTimeObserver];
    self.playerTimeObserver = nil;
  }

  NSString *finishedSegmentId = self.currentSegmentId;

  dispatch_async(dispatch_get_main_queue(), ^{
    if ([self.delegate
            respondsToSelector:@selector(pipelineDidFinishSegment:)]) {
      [self.delegate pipelineDidFinishSegment:finishedSegmentId];
    }
  });

  dispatch_async(self.playbackQueue, ^{
    [self playNextSegment];
  });
}

#pragma mark - AudioStreamPlayerDelegate

- (void)audioStreamPlayerDidStartSegment:(NSString *)segmentId {
  dispatch_async(dispatch_get_main_queue(), ^{
    if ([self.delegate
            respondsToSelector:@selector(pipelineDidStartSegment:duration:)]) {
      [self.delegate pipelineDidStartSegment:segmentId duration:0];
    }
  });
}

- (void)audioStreamPlayerDidUpdateTime:(NSTimeInterval)time
                             segmentId:(NSString *)segmentId {
  dispatch_async(dispatch_get_main_queue(), ^{
    if ([self.delegate respondsToSelector:@selector
                       (pipelineDidUpdatePlaybackTime:segmentId:)]) {
      [self.delegate pipelineDidUpdatePlaybackTime:time segmentId:segmentId];
    }
  });
}

- (void)audioStreamPlayerDidFinishSegment:(NSString *)segmentId {
  dispatch_async(dispatch_get_main_queue(), ^{
    if ([self.delegate
            respondsToSelector:@selector(pipelineDidFinishSegment:)]) {
      [self.delegate pipelineDidFinishSegment:segmentId];
    }
  });

  dispatch_async(self.playbackQueue, ^{
    // 检查是否还有更多片段
    if (self.segmentQueue.count == 0) {
      self.playing = NO;
      self.currentSegmentId = nil;

      dispatch_async(dispatch_get_main_queue(), ^{
        if ([self.delegate
                respondsToSelector:@selector(pipelineDidFinishAllSegments)]) {
          [self.delegate pipelineDidFinishAllSegments];
        }
      });
    }
  });
}

@end