使用自定义 StoreLayout

package nexus.io.telegram.bots.service;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.BiFunction;

import com.fasterxml.jackson.core.type.TypeReference;
import nexus.io.db.activerecord.Db;
import nexus.io.db.activerecord.Row;
import nexus.io.telegram.store.ChannelInfo;
import nexus.io.telegram.store.TlObjectMapperUtils;
import nexus.io.telegram.store.service.DbChannelStoreService;

import lombok.extern.slf4j.Slf4j;
import telegram4j.tl.ChannelParticipant;
import telegram4j.tl.ImmutableChannel;
import telegram4j.tl.ImmutableChannelFull;
import telegram4j.tl.Peer;

@Slf4j
public class DbChannelStoreServiceImpl implements DbChannelStoreService {
  public static final String TELEGRAM_CHANNEL_MIN = "telegram_channel_min";
  public static final String TELEGRAM_CHANNEL_FULL = "telegram_channel_full";
  public static final String TELEGRAM_CHANNEL_PARTICIPANTS = "telegram_channel_participants";

  private TypeReference<ConcurrentMap<Peer, ChannelParticipant>> typeRef = new TypeReference<ConcurrentMap<Peer, ChannelParticipant>>() {
  };

  private final ConcurrentMap<Long, ReentrantLock> lockMap = new ConcurrentHashMap<>();

  /**
   * 从数据库中获取 ChannelInfo
   *
   * @param rawPeerId 频道的原始 Peer ID
   * @return ChannelInfo 对象，如果不存在则返回 null
   */
  @Override
  public ChannelInfo get(Long rawPeerId) {
    ChannelInfo info = null;

    // 获取基本信息
    Row row = Db.findById(TELEGRAM_CHANNEL_MIN, rawPeerId);
    if (row != null) {
      String str = row.getStr("json");
      try {
        ImmutableChannel min = TlObjectMapperUtils.readValue(str, ImmutableChannel.class);
        info = new ChannelInfo(min);
      } catch (Exception e) {
        log.error("Failed to readValue: rawPeerId={}, json={}, error={}", rawPeerId, str, e.getMessage(), e);
        return null;
      }
    } else {
      return null;
    }

    // 获取详细信息
    row = Db.findById(TELEGRAM_CHANNEL_FULL, rawPeerId);
    if (row != null) {
      String str = row.getStr("json");
      try {
        ImmutableChannelFull full = TlObjectMapperUtils.readValue(str, ImmutableChannelFull.class);
        info.setFull(full);
      } catch (Exception e) {
        log.error("Failed to readValue: rawPeerId={}, json={}, error={}", rawPeerId, str, e.getMessage(), e);
        return null;
      }
    } else {
      return null;
    }

    // 获取参与者信息
    row = Db.findById(TELEGRAM_CHANNEL_PARTICIPANTS, rawPeerId);
    if (row != null) {
      String str = row.getStr("json");
      try {
        ConcurrentMap<Peer, ChannelParticipant> participants = TlObjectMapperUtils.readValue(str, typeRef);
        info.setParticipants(participants);
      } catch (Exception e) {
        log.error("Failed to readValue: rawPeerId={}, json={}, error={}", rawPeerId, str, e.getMessage(), e);
      }
    }

    return info;
  }

  /**
   * 如果频道存在，则计算新的 ChannelInfo 并更新数据库
   *
   * @param channelId         频道 ID
   * @param remappingFunction 计算函数
   * @return 更新后的 ChannelInfo
   */
  @Override
  public ChannelInfo computeIfPresent(Long channelId, BiFunction<? super Long, ? super ChannelInfo, ? extends ChannelInfo> remappingFunction) {
    ChannelInfo existing = get(channelId);
    if (existing != null) {
      ChannelInfo updated = remappingFunction.apply(channelId, existing);
      if (updated != null) {
        updateChannelInfo(channelId, updated);
        return updated;
      } else {
        deleteChannelInfo(channelId);
      }
    }
    return null;
  }

  /**
   * 计算并更新或插入 ChannelInfo
   *
   * @param id                 频道 ID
   * @param remappingFunction  计算函数
   */
  @Override
  public void compute(Long id, BiFunction<? super Long, ? super ChannelInfo, ? extends ChannelInfo> remappingFunction) {
    ChannelInfo existing = get(id);
    ChannelInfo updated = remappingFunction.apply(id, existing);

    if (updated != null) {
      if (existing != null) {
        updateChannelInfo(id, updated);
      } else {
        insertChannelInfo(id, updated);
      }
    } else {
      if (existing != null) {
        deleteChannelInfo(id);
      }
    }
  }

  /**
   * 插入或更新 ChannelInfo
   *
   * @param channelId 频道 ID
   * @param info      ChannelInfo 对象
   */
  private void insertChannelInfo(Long channelId, ChannelInfo info) {
    ReentrantLock lock = lockMap.computeIfAbsent(channelId, id -> new ReentrantLock());

    // 处理基本信息
    if (info.getMin() != null) {
      if (info.getMin() instanceof ImmutableChannel immutableChannel) {
        String json = TlObjectMapperUtils.writeValueAsString(immutableChannel);
        lock.lock();
        try {
          Row row = Row.by("id", channelId).set("json", json);
          if (Db.exists(TELEGRAM_CHANNEL_MIN, "id", channelId)) {
            Db.update(TELEGRAM_CHANNEL_MIN, row);
          } else {
            Db.save(TELEGRAM_CHANNEL_MIN, row);
          }
        } finally {
          lock.unlock();
        }
      } else {
        log.error("Unsupported class type for min: {}", info.getMin().getClass());
      }
    }

    // 处理详细信息
    if (info.getFull() != null) {
      String json = TlObjectMapperUtils.writeValueAsString(info.getFull());
      Row fullRow = Row.by("id", channelId).set("json", json);
      lock.lock();
      try {
        if (Db.exists(TELEGRAM_CHANNEL_FULL, "id", channelId)) {
          Db.update(TELEGRAM_CHANNEL_FULL, fullRow);
        } else {
          Db.save(TELEGRAM_CHANNEL_FULL, fullRow);
        }
      } finally {
        lock.unlock();
      }
    }

    // 处理参与者信息
    if (info.getParticipants() != null) {
      String json = TlObjectMapperUtils.writeValueAsString(info.getParticipants());
      Row row = Row.by("id", channelId).set("json", json);
      lock.lock();
      try {
        if (Db.exists(TELEGRAM_CHANNEL_PARTICIPANTS, "id", channelId)) {
          Db.update(TELEGRAM_CHANNEL_PARTICIPANTS, row);
        } else {
          Db.save(TELEGRAM_CHANNEL_PARTICIPANTS, row);
        }
      } finally {
        lock.unlock();
      }
    }
  }

  /**
   * 更新现有的 ChannelInfo
   *
   * @param channelId 频道 ID
   * @param info      更新后的 ChannelInfo 对象
   */
  private void updateChannelInfo(Long channelId, ChannelInfo info) {
    insertChannelInfo(channelId, info);
  }

  /**
   * 删除 ChannelInfo
   *
   * @param channelId 频道 ID
   */
  private void deleteChannelInfo(Long channelId) {
    String deleteMinSql = "DELETE FROM " + TELEGRAM_CHANNEL_MIN + " WHERE id = ?";
    String deleteFullSql = "DELETE FROM " + TELEGRAM_CHANNEL_FULL + " WHERE id = ?";
    String deleteParticipantsSql = "DELETE FROM " + TELEGRAM_CHANNEL_PARTICIPANTS + " WHERE id = ?";
    Db.delete(deleteMinSql, channelId);
    Db.delete(deleteFullSql, channelId);
    Db.delete(deleteParticipantsSql, channelId);
  }
}

// 初始化 DbChannelStoreServiceImpl
DbChannelStoreServiceImpl dbChannelStoreService = new DbChannelStoreServiceImpl();

// 创建 DbStoreLayoutImpl，并传入必要的函数
DbStoreLayoutImpl storeLayoutImpl = new DbStoreLayoutImpl(Function.identity());

// 设置自定义的 DbChannelStoreService
storeLayoutImpl.setDbChannels(dbChannelStoreService);

// 创建 FileStoreLayout，并指定存储路径（例如：t4j-bot_{botId}.bin）
FileStoreLayout storeLayout = new FileStoreLayout(storeLayoutImpl, Path.of("t4j-bot_" + botId + ".bin"));

// 配置引导程序使用自定义的 StoreLayout
bootstrap.setStoreLayout(storeLayout);