back/src/modules/workers/workers.service.ts - master - guselnikov/inst

back/src/modules/workers/workers.service.ts
1 305 строк47 KB
Viktor Guselnikov
new
26 июл 2026, 18:42
26 июл 2026, 18:42322e1f1
import { Injectable, Logger } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { InjectRepository } from '@nestjs/typeorm';
import { Repository } from 'typeorm';
import { IgAccountStatus, ParsingJobStatus, FilterRunStatus, EnrichRunStatus, VatagoSyncRunStatus, VatagoSyncMemberStatus, QueueName, LikeJobStatus, LikeJobMode, IgPostStatus } from '../../common/enums';
import { normalizePhoneForVatago } from '../../common/utils/vatago-phone.util';
import { decryptSecret } from '../../common/utils/crypto.util';
import {
  generateBrowserFingerprint,
  type BrowserFingerprint,
} from '../../common/utils/browser-fingerprint.util';
import { collectedProfileMatchesFilter } from '../../common/utils/profile-filter';
import { CollectedProfile } from '../../database/entities/collected-profile.entity';
import { EnrichRun } from '../../database/entities/enrich-run.entity';
import { VatagoSyncRun } from '../../database/entities/vatago-sync-run.entity';
import { VatagoSyncResult } from '../../database/entities/vatago-sync-result.entity';
import { FilterRun } from '../../database/entities/filter-run.entity';
import { FilteredResult } from '../../database/entities/filtered-result.entity';
import { IgAccount } from '../../database/entities/ig-account.entity';
import { ParsingJob } from '../../database/entities/parsing-job.entity';
import { LikeJob } from '../../database/entities/like-job.entity';
import { IgPost } from '../../database/entities/ig-post.entity';
import { InstagramScraperService } from '../instagram/instagram-scraper.service';
import { IgRemoteLoginService } from '../instagram/ig-remote-login.service';
import { IgBrowserLogService } from '../instagram/ig-browser-log.service';
import { CollectedProfilesService } from '../collected-profiles/collected-profiles.service';
import { AudiencesService } from '../audiences/audiences.service';
import { VatagoClientService } from '../vatago/vatago-client.service';
import { VatagoSyncService } from '../vatago-sync/vatago-sync.service';
import { ParsingLogService, type ParsingLogLevel } from '../parsing-jobs/parsing-log.service';
import { LikeLogService, type LikeLogLevel } from '../like-jobs/like-log.service';
import { IgPostLogService, type IgPostLogLevel } from '../ig-posts/ig-post-log.service';
import { QueueService } from '../queue/queue.service';
import { AiOrchestratorService } from '../ai-studio/ai-orchestrator.service';
import { RedisService } from '../redis/redis.service';
import { EventsGateway } from '../websocket/events.gateway';

@Injectable()
export class WorkersService {
  private readonly logger = new Logger(WorkersService.name);

  constructor(
    private readonly queue: QueueService,
    @InjectRepository(IgAccount)
    private readonly igAccounts: Repository<IgAccount>,
    @InjectRepository(ParsingJob)
    private readonly parsingJobs: Repository<ParsingJob>,
    @InjectRepository(LikeJob)
    private readonly likeJobs: Repository<LikeJob>,
    @InjectRepository(IgPost)
    private readonly igPosts: Repository<IgPost>,
    @InjectRepository(FilterRun)
    private readonly filterRuns: Repository<FilterRun>,
    @InjectRepository(EnrichRun)
    private readonly enrichRuns: Repository<EnrichRun>,
    @InjectRepository(VatagoSyncRun)
    private readonly vatagoSyncRuns: Repository<VatagoSyncRun>,
    @InjectRepository(VatagoSyncResult)
    private readonly vatagoSyncResults: Repository<VatagoSyncResult>,
    @InjectRepository(FilteredResult)
    private readonly filteredResults: Repository<FilteredResult>,
    @InjectRepository(CollectedProfile)
    private readonly collectedProfiles: Repository<CollectedProfile>,
    private readonly scraper: InstagramScraperService,
    private readonly remoteLogin: IgRemoteLoginService,
    private readonly browserLog: IgBrowserLogService,
    private readonly parsingLog: ParsingLogService,
    private readonly likeLog: LikeLogService,
    private readonly postLog: IgPostLogService,
    private readonly profilesService: CollectedProfilesService,
    private readonly audiencesService: AudiencesService,
    private readonly vatagoClient: VatagoClientService,
    private readonly vatagoSyncService: VatagoSyncService,
    private readonly redis: RedisService,
    private readonly events: EventsGateway,
    private readonly config: ConfigService,
    private readonly aiOrchestrator: AiOrchestratorService,
  ) {
    this.queue.registerHandler(QueueName.IG_LOGIN, (payload) => this.handleIgLogin(payload));
    this.queue.registerHandler(QueueName.PARSING_RUN, (payload) => this.handleParsing(payload));
    this.queue.registerHandler(QueueName.FILTER_RUN, (payload) => this.handleFilter(payload));
    this.queue.registerHandler(QueueName.ENRICH_RUN, (payload) => this.handleEnrich(payload));
    this.queue.registerHandler(QueueName.VATAGO_SYNC, (payload) => this.handleVatagoSync(payload));
    this.queue.registerHandler(QueueName.LIKE_RUN, (payload) => this.handleLike(payload));
    this.queue.registerHandler(QueueName.IG_POST_PUBLISH, (payload) => this.handleIgPostPublish(payload));
    this.queue.registerHandler(QueueName.AI_RUN, (payload) => this.handleAiRun(payload));
  }

  private async handleAiRun(raw: string): Promise<void> {
    const { runId } = JSON.parse(raw) as { runId: string };
    await this.aiOrchestrator.executeRun(runId);
  }

  private async handleIgLogin(raw: string): Promise<void> {
    const { igAccountId, confirmManual } = JSON.parse(raw) as {
      igAccountId: string;
      confirmManual?: boolean;
    };
    const account = await this.igAccounts.findOne({ where: { id: igAccountId } });
    if (!account) return;

    this.browserLog.log(igAccountId, 'worker.ig_login.received', {
      confirmManual: Boolean(confirmManual),
      username: account.username,
      status: account.status,
    });

    if (confirmManual) {
      try {
        const storageState = await this.scraper.saveManualSession(igAccountId);
        this.remoteLogin.stopManualSession(igAccountId);
        account.storageState = storageState;
        account.status = IgAccountStatus.ACTIVE;
        await this.igAccounts.save(account);
        this.browserLog.log(igAccountId, 'worker.ig_login.confirm.success', {});
        this.events.emitIg(igAccountId, 'ig.login.active', { igAccountId });
      } catch (error) {
        this.remoteLogin.stopManualSession(igAccountId);
        account.status = IgAccountStatus.EXPIRED;
        await this.igAccounts.save(account);
        this.browserLog.log(igAccountId, 'worker.ig_login.confirm.failed', {
          message: (error as Error).message,
        });
        this.events.emitIg(igAccountId, 'ig.login.failed', {
          igAccountId,
          code: 'browser_closed',
          message:
            'Браузер закрыт или вход не завершён. Нажми «Повторить вход» — откроется новый браузер.',
        });
      }
      return;
    }

    this.remoteLogin.stopManualSession(igAccountId);

    try {
      this.events.emitIg(igAccountId, 'ig.login.started', { igAccountId });
      if (!account.browserFingerprint) {
        account.browserFingerprint = generateBrowserFingerprint(account.id) as unknown as Record<
          string,
          unknown
        >;
        await this.igAccounts.save(account);
      }
      const fingerprint = account.browserFingerprint as unknown as BrowserFingerprint;
      const password = decryptSecret(
        account.passwordEncrypted!,
        this.config.get<string>('encryption.key')!,
      );
      const result = await this.scraper.login(igAccountId, account.username, password, fingerprint, {
        onCredentialsFilled: () => {
          this.events.emitIg(igAccountId, 'ig.login.credentials_filled', { igAccountId });
        },
      });

      if (result.status === 'awaiting_manual') {
        account.status = IgAccountStatus.AWAITING_MANUAL;
        await this.igAccounts.save(account);
        this.browserLog.log(igAccountId, 'worker.ig_login.awaiting_manual', {
          reason: result.reason,
        });
        this.remoteLogin.startManualSession(igAccountId);
        this.events.emitIg(igAccountId, 'ig.awaiting_manual', {
          igAccountId,
          reason: result.reason,
          message: result.message,
          remote: true,
        });
        return;
      }

      account.storageState = result.storageState;
      account.status = IgAccountStatus.ACTIVE;
      await this.igAccounts.save(account);
      this.browserLog.log(igAccountId, 'worker.ig_login.active', {});
      this.events.emitIg(igAccountId, 'ig.login.active', { igAccountId });
    } catch (error) {
      this.remoteLogin.stopManualSession(igAccountId);
      account.status = IgAccountStatus.EXPIRED;
      await this.igAccounts.save(account);
      this.browserLog.log(igAccountId, 'worker.ig_login.failed', {
        message: (error as Error).message,
      });
      this.events.emitIg(igAccountId, 'ig.login.failed', {
        igAccountId,
        code: 'login_failed',
        message: (error as Error).message,
      });
    }
  }

  private async handleParsing(raw: string): Promise<void> {
    const { parsingJobId } = JSON.parse(raw) as { parsingJobId: string };
    const job = await this.parsingJobs.findOne({
      where: { id: parsingJobId },
      relations: { igAccount: true },
    });
    if (!job || !job.igAccount?.storageState) {
      this.pLog(parsingJobId, 'warn', 'job.skip', { reason: 'missing_job_or_session' });
      return;
    }
    if (job.status !== ParsingJobStatus.RUNNING) {
      this.pLog(parsingJobId, 'warn', 'job.skip', { reason: 'not_running', status: job.status });
      return;
    }

    const lockKey = `lock:ig:${job.igAccountId}`;
    const locked = await this.acquireIgLock(job.igAccountId, parsingJobId);
    if (!locked) {
      await this.revertJobOnLockFailure(job);
      return;
    }

    const isRunning = () => this.isParsingRunning(parsingJobId);

    try {
      this.pLog(parsingJobId, 'info', 'job.started', {
        hashtag: job.hashtag,
        maxProfiles: job.maxProfiles,
        collectPhone: job.collectPhone,
        collectAvatar: job.collectAvatar,
        recentPostsCount: job.recentPostsCount,
        audienceId: job.audienceId,
        igAccount: job.igAccount.username,
      });
      this.events.emitParsing(parsingJobId, 'parsing.started', { parsingJobId });

      const maxScrolls = this.config.get<number>('instagram.maxScrolls', 100);
      const storagePath = this.config.get<string>('storage.path', '../storage');

      const alreadyScraped = await this.getScrapedUsernames(parsingJobId);
      const scrapeGoal = Math.max(0, job.maxProfiles - alreadyScraped.size);
      this.pLog(parsingJobId, 'info', 'discovery.start', {
        maxScrolls,
        alreadyScraped: alreadyScraped.size,
        maxProfiles: job.maxProfiles,
        scrapeGoal,
      });

      let profilesScraped = job.profilesScraped;
      let progressErrors = 0;
      let progressSkipped = 0;

      const parseResult = await this.scraper.executeHashtagParsing(
        job.igAccount.storageState,
        {
          hashtag: job.hashtag,
          maxProfiles: job.maxProfiles,
          maxScrolls,
          collectPhone: job.collectPhone,
          collectAvatar: job.collectAvatar,
          recentPostsCount: job.recentPostsCount,
          storagePath,
          alreadyScraped,
          excludeUsername: job.igAccount.username,
        },
        {
          shouldContinue: isRunning,
          onDiscoveryProgress: (found) => {
            job.profilesFound = found;
            void this.parsingJobs.save(job);
            this.pLog(parsingJobId, 'debug', 'discovery.progress', { found, max: job.maxProfiles });
            this.events.emitParsing(parsingJobId, 'parsing.progress', {
              parsingJobId,
              phase: 'discovery',
              scraped: profilesScraped,
              found,
              max: job.maxProfiles,
            });
          },
          onLog: (event, data) => {
            const level: ParsingLogLevel =
              event.includes('failed') || event === 'search.fallback_direct_url'
                ? 'warn'
                : event.includes('success') || event.includes('done') || event.includes('phase')
                  ? 'info'
                  : 'debug';
            this.pLog(parsingJobId, level, event, data);

            if (event === 'discovery.phase.done' && data) {
              const total = Number(data.total ?? 0);
              job.profilesFound = total;
              void this.parsingJobs.save(job);
              this.events.emitParsing(parsingJobId, 'parsing.discovery_done', {
                parsingJobId,
                found: total,
                usernames: data.usernames ?? [],
              });
            }
            if (event === 'scrape.phase.start' && data) {
              this.events.emitParsing(parsingJobId, 'parsing.progress', {
                parsingJobId,
                phase: 'scrape',
                scraped: profilesScraped,
                found: job.profilesFound,
                max: job.maxProfiles,
                totalToScrape: data.count,
              });
            }
          },
          onProfileStart: (username) => {
            this.pLog(parsingJobId, 'info', 'profile.start', { username });
          },
          onProfileSuccess: async (scraped) => {
            const profile = await this.profilesService.upsertProfile({
              organizationId: job.organizationId,
              parsingJobId: job.id,
              username: scraped.username,
              displayName: scraped.displayName,
              bio: scraped.bio,
              phone: scraped.phone,
              phoneSource: scraped.phoneSource,
              isBusiness: scraped.isBusiness,
              avatarPath: scraped.avatarPath,
              avatarHash: scraped.avatarHash,
              sourceHashtag: job.hashtag,
              scrapedAt: new Date(),
              rawMeta: scraped.rawMeta,
            });
            await this.profilesService.syncMedia(profile.id, scraped.recentPosts);
            if (job.audienceId) {
              await this.audiencesService.upsertFromParsing(
                job.organizationId,
                job.audienceId,
                profile,
                scraped.recentPosts,
              );
            }
            profilesScraped += 1;
            job.profilesScraped = profilesScraped;
            await this.parsingJobs.save(job);
            this.events.emitParsing(parsingJobId, 'parsing.progress', {
              parsingJobId,
              phase: 'scrape',
              scraped: profilesScraped,
              found: job.profilesFound,
              max: job.maxProfiles,
              currentUsername: scraped.username,
              errors: progressErrors,
              skipped: progressSkipped,
            });
          },
          onProfileError: (username, message) => {
            progressErrors += 1;
            this.pLog(parsingJobId, 'error', 'profile.failed', { username, message });
            this.events.emitParsing(parsingJobId, 'parsing.progress', {
              parsingJobId,
              phase: 'scrape',
              scraped: profilesScraped,
              found: job.profilesFound,
              max: job.maxProfiles,
              currentUsername: username,
              errors: progressErrors,
              skipped: progressSkipped,
              lastError: message,
            });
          },
          sleep: (ms) => this.scraper.sleep(ms),
          randomDelay: () => this.scraper.randomDelay(),
        },
      );

      progressErrors = parseResult.errors;
      progressSkipped = parseResult.skipped;

      if (!(await isRunning())) {
        await this.finishPaused(parsingJobId);
        return;
      }

      job.profilesFound = parseResult.usernames.length;
      job.profilesScraped = profilesScraped;
      await this.parsingJobs.save(job);
      this.pLog(parsingJobId, 'info', 'discovery.done', {
        total: parseResult.usernames.length,
        errors: parseResult.errors,
        skipped: parseResult.skipped,
      });

      const fresh = await this.parsingJobs.findOne({ where: { id: parsingJobId } });
      if (fresh?.status !== ParsingJobStatus.RUNNING) {
        await this.finishPaused(parsingJobId);
        return;
      }

      fresh.status = ParsingJobStatus.COMPLETED;
      fresh.completedAt = new Date();
      fresh.profilesScraped = profilesScraped;
      await this.parsingJobs.save(fresh);
      this.pLog(parsingJobId, 'info', 'job.completed', {
        profilesScraped,
        profilesFound: fresh.profilesFound,
        errors: parseResult.errors,
        skipped: parseResult.skipped,
      });
      this.events.emitParsing(parsingJobId, 'parsing.completed', {
        parsingJobId,
        profilesScraped,
        errors: parseResult.errors,
        skipped: parseResult.skipped,
      });
    } catch (error) {
      const message = (error as Error).message;
      job.status = ParsingJobStatus.FAILED;
      job.errorMessage = message;
      await this.parsingJobs.save(job);
      this.pLog(parsingJobId, 'error', 'job.failed', { message, stack: (error as Error).stack });
      this.events.emitParsing(parsingJobId, 'parsing.failed', {
        parsingJobId,
        message,
      });
    } finally {
      await this.redis.releaseLock(lockKey);
      this.pLog(parsingJobId, 'debug', 'job.lock_released', { lockKey });
    }
  }

  private async acquireIgLock(igAccountId: string, parsingJobId: string): Promise<boolean> {
    const lockKey = `lock:ig:${igAccountId}`;
    const ttlMs = 60 * 60 * 1000;

    if (await this.redis.acquireLock(lockKey, ttlMs)) {
      this.pLog(parsingJobId, 'debug', 'lock.acquired', { lockKey });
      return true;
    }

    const running = await this.parsingJobs.find({
      where: { igAccountId, status: ParsingJobStatus.RUNNING },
    });
    const otherRunning = running.filter((item) => item.id !== parsingJobId);

    const runningLike = await this.likeJobs.find({
      where: { igAccountId, status: LikeJobStatus.RUNNING },
    });

    if (otherRunning.length > 0) {
      this.pLog(parsingJobId, 'warn', 'job.skip', {
        reason: 'ig_account_busy',
        igAccountId,
        otherJobId: otherRunning[0].id,
        otherHashtag: otherRunning[0].hashtag,
      });
      return false;
    }

    if (runningLike.length > 0) {
      this.pLog(parsingJobId, 'warn', 'job.skip', {
        reason: 'ig_account_busy_like',
        igAccountId,
        otherJobId: runningLike[0].id,
      });
      return false;
    }

    await this.redis.releaseLock(lockKey);
    this.pLog(parsingJobId, 'info', 'lock.stale_released', { lockKey });

    if (await this.redis.acquireLock(lockKey, ttlMs)) {
      this.pLog(parsingJobId, 'debug', 'lock.acquired', { lockKey, afterStaleRelease: true });
      return true;
    }

    this.pLog(parsingJobId, 'warn', 'job.skip', { reason: 'ig_account_busy', igAccountId });
    return false;
  }

  private async revertJobOnLockFailure(job: ParsingJob): Promise<void> {
    job.status = ParsingJobStatus.PAUSED;
    job.errorMessage = 'IG аккаунт занят. Если другой job не идёт — нажми «Продолжить» ещё раз.';
    await this.parsingJobs.save(job);
    this.events.emitParsing(job.id, 'parsing.failed', {
      parsingJobId: job.id,
      code: 'ig_account_busy',
      message: job.errorMessage,
    });
  }

  private async isParsingRunning(parsingJobId: string): Promise<boolean> {
    const job = await this.parsingJobs.findOne({ where: { id: parsingJobId } });
    return job?.status === ParsingJobStatus.RUNNING;
  }

  private async finishPaused(parsingJobId: string): Promise<void> {
    const job = await this.parsingJobs.findOne({ where: { id: parsingJobId } });
    if (!job) return;
    if (job.status !== ParsingJobStatus.PAUSED) {
      job.status = ParsingJobStatus.PAUSED;
      await this.parsingJobs.save(job);
    }
    this.pLog(parsingJobId, 'info', 'job.paused', {
      profilesScraped: job.profilesScraped,
      profilesFound: job.profilesFound,
    });
    this.events.emitParsing(parsingJobId, 'parsing.paused', {
      parsingJobId,
      profilesScraped: job.profilesScraped,
      profilesFound: job.profilesFound,
    });
  }

  private async getScrapedUsernames(parsingJobId: string): Promise<Set<string>> {
    const rows = await this.collectedProfiles.find({
      where: { parsingJobId },
      select: { username: true },
    });
    return new Set(rows.map((row) => row.username.toLowerCase()));
  }

  private pLog(
    parsingJobId: string,
    level: ParsingLogLevel,
    event: string,
    data?: Record<string, unknown>,
  ): void {
    this.parsingLog.log(parsingJobId, level, event, data);
    this.events.emitParsing(parsingJobId, 'parsing.log', {
      parsingJobId,
      level,
      event,
      ts: new Date().toISOString(),
      ...data,
    });
  }

  private async handleLike(raw: string): Promise<void> {
    const { likeJobId } = JSON.parse(raw) as { likeJobId: string };
    const job = await this.likeJobs.findOne({
      where: { id: likeJobId },
      relations: { igAccount: true },
    });
    if (!job || !job.igAccount?.storageState) {
      this.lLike(likeJobId, 'warn', 'job.skip', { reason: 'missing_job_or_session' });
      return;
    }
    if (job.status !== LikeJobStatus.RUNNING) {
      this.lLike(likeJobId, 'warn', 'job.skip', { reason: 'not_running', status: job.status });
      return;
    }

    const lockKey = `lock:ig:${job.igAccountId}`;
    const locked = await this.acquireLikeLock(job.igAccountId, likeJobId);
    if (!locked) {
      await this.revertLikeOnLockFailure(job);
      return;
    }

    const isRunning = () => this.isLikeRunning(likeJobId);

    try {
      this.lLike(likeJobId, 'info', 'job.started', {
        mode: job.mode,
        maxLikes: job.maxLikes,
        delayMinMs: job.delayMinMs,
        delayMaxMs: job.delayMaxMs,
        likesPerProfile: job.likesPerProfile,
        followEnabled: job.followEnabled,
        accountDelayMinMs: job.accountDelayMinMs,
        accountDelayMaxMs: job.accountDelayMaxMs,
        igAccount: job.igAccount.username,
      });
      this.events.emitLike(likeJobId, 'like.started', { likeJobId });

      const maxScrolls = this.config.get<number>('instagram.maxScrolls', 100);

      if (job.mode === LikeJobMode.AUDIENCE) {
        if (!job.audienceId) {
          throw new Error('Audience like job missing audienceId');
        }

        const usernames = await this.audiencesService.listMemberUsernames(
          job.audienceId,
          job.organizationId,
        );
        if (usernames.length === 0) {
          throw new Error('Audience has no members');
        }

        if (job.profilesTotal === 0) {
          job.profilesTotal = usernames.length;
          job.maxLikes = job.likesPerProfile * usernames.length;
          await this.likeJobs.save(job);
        }

        const result = await this.scraper.executeAudienceLiking(
          job.igAccount.storageState,
          {
            usernames,
            likesPerProfile: job.likesPerProfile,
            followEnabled: job.followEnabled,
            likeDelayMinMs: job.delayMinMs,
            likeDelayMaxMs: job.delayMaxMs,
            accountDelayMinMs: job.accountDelayMinMs,
            accountDelayMaxMs: job.accountDelayMaxMs,
            maxScrolls,
            startProfileIndex: job.profilesDone,
          },
          {
            shouldContinue: isRunning,
            onLog: (event, data) => {
              const level: LikeLogLevel =
                event.includes('error') || event.includes('failed')
                  ? 'error'
                  : event.includes('skip') || event.includes('stagnant') || event.includes('exhausted')
                    ? 'warn'
                    : event.includes('success') || event.includes('started') || event.includes('opened')
                      ? 'info'
                      : 'debug';
              this.lLike(likeJobId, level, event, data);
            },
            onLikeSuccess: async () => {
              job.likesDone += 1;
              await this.likeJobs.save(job);
              this.events.emitLike(likeJobId, 'like.progress', {
                likeJobId,
                likesDone: job.likesDone,
                likesSkipped: job.likesSkipped,
                maxLikes: job.maxLikes,
                profilesDone: job.profilesDone,
                profilesTotal: job.profilesTotal,
                errors: job.errorsCount,
              });
            },
            onLikeSkip: () => {
              job.likesSkipped += 1;
              void this.likeJobs.save(job);
            },
            onLikeError: () => {
              job.errorsCount += 1;
              void this.likeJobs.save(job);
            },
            onProfileStart: async (username) => {
              this.events.emitLike(likeJobId, 'like.progress', {
                likeJobId,
                currentUsername: username,
                likesDone: job.likesDone,
                likesSkipped: job.likesSkipped,
                profilesDone: job.profilesDone,
                profilesTotal: job.profilesTotal,
                maxLikes: job.maxLikes,
                errors: job.errorsCount,
              });
            },
            onProfileDone: async () => {
              job.profilesDone += 1;
              await this.likeJobs.save(job);
              this.events.emitLike(likeJobId, 'like.progress', {
                likeJobId,
                likesDone: job.likesDone,
                likesSkipped: job.likesSkipped,
                profilesDone: job.profilesDone,
                profilesTotal: job.profilesTotal,
                maxLikes: job.maxLikes,
                errors: job.errorsCount,
              });
            },
            sleep: (ms) => this.scraper.sleep(ms),
            randomDelay: (minMs, maxMs) => this.scraper.randomDelay(minMs, maxMs),
          },
        );

        if (!(await isRunning())) {
          await this.finishLikePaused(likeJobId);
          return;
        }

        job.likesDone = result.liked;
        job.likesSkipped = result.skipped;
        job.errorsCount = result.errors;
        await this.likeJobs.save(job);

        const freshAudience = await this.likeJobs.findOne({ where: { id: likeJobId } });
        if (freshAudience?.status !== LikeJobStatus.RUNNING) {
          await this.finishLikePaused(likeJobId);
          return;
        }

        freshAudience.status = LikeJobStatus.COMPLETED;
        freshAudience.completedAt = new Date();
        await this.likeJobs.save(freshAudience);
        this.lLike(likeJobId, 'info', 'job.completed', {
          likesDone: result.liked,
          likesSkipped: result.skipped,
          errors: result.errors,
          profilesProcessed: result.profilesProcessed,
          profilesSkipped: result.profilesSkipped,
        });
        this.events.emitLike(likeJobId, 'like.completed', {
          likeJobId,
          likesDone: result.liked,
          likesSkipped: result.skipped,
          errors: result.errors,
          profilesDone: freshAudience.profilesDone,
          profilesTotal: freshAudience.profilesTotal,
        });
        return;
      }

      const storagePath = this.config.get<string>('storage.path', '../storage');

      const result = await this.scraper.executeFeedLiking(
        job.igAccount.storageState,
        {
          maxLikes: job.maxLikes,
          maxScrolls,
          delayMinMs: job.delayMinMs,
          delayMaxMs: job.delayMaxMs,
          storagePath,
        },
        {
          shouldContinue: isRunning,
          onLog: (event, data) => {
            const level: LikeLogLevel =
              event.includes('error') || event.includes('failed')
                ? 'error'
                : event.includes('skip') || event.includes('stagnant')
                  ? 'warn'
                  : event.includes('success') || event.includes('started') || event.includes('opened')
                    ? 'info'
                    : 'debug';
            this.lLike(likeJobId, level, event, data);
          },
          onLikeSuccess: async () => {
            job.likesDone += 1;
            await this.likeJobs.save(job);
            this.events.emitLike(likeJobId, 'like.progress', {
              likeJobId,
              likesDone: job.likesDone,
              likesSkipped: job.likesSkipped,
              maxLikes: job.maxLikes,
              errors: job.errorsCount,
            });
          },
          onLikeSkip: () => {
            job.likesSkipped += 1;
            void this.likeJobs.save(job);
          },
          onLikeError: () => {
            job.errorsCount += 1;
            void this.likeJobs.save(job);
          },
          sleep: (ms) => this.scraper.sleep(ms),
          randomDelay: (minMs, maxMs) => this.scraper.randomDelay(minMs, maxMs),
        },
      );

      if (!(await isRunning())) {
        await this.finishLikePaused(likeJobId);
        return;
      }

      job.likesDone = result.liked;
      job.likesSkipped = result.skipped;
      job.errorsCount = result.errors;
      await this.likeJobs.save(job);

      const fresh = await this.likeJobs.findOne({ where: { id: likeJobId } });
      if (fresh?.status !== LikeJobStatus.RUNNING) {
        await this.finishLikePaused(likeJobId);
        return;
      }

      fresh.status = LikeJobStatus.COMPLETED;
      fresh.completedAt = new Date();
      await this.likeJobs.save(fresh);
      this.lLike(likeJobId, 'info', 'job.completed', {
        likesDone: result.liked,
        likesSkipped: result.skipped,
        errors: result.errors,
      });
      this.events.emitLike(likeJobId, 'like.completed', {
        likeJobId,
        likesDone: result.liked,
        likesSkipped: result.skipped,
        errors: result.errors,
      });
    } catch (error) {
      const message = (error as Error).message;
      job.status = LikeJobStatus.FAILED;
      job.errorMessage = message;
      await this.likeJobs.save(job);
      this.lLike(likeJobId, 'error', 'job.failed', { message, stack: (error as Error).stack });
      this.events.emitLike(likeJobId, 'like.failed', { likeJobId, message });
    } finally {
      await this.redis.releaseLock(lockKey);
      this.lLike(likeJobId, 'debug', 'job.lock_released', { lockKey });
    }
  }

  private async acquireLikeLock(igAccountId: string, likeJobId: string): Promise<boolean> {
    const lockKey = `lock:ig:${igAccountId}`;
    const ttlMs = 60 * 60 * 1000;

    if (await this.redis.acquireLock(lockKey, ttlMs)) {
      this.lLike(likeJobId, 'debug', 'lock.acquired', { lockKey });
      return true;
    }

    const runningLike = await this.likeJobs.find({
      where: { igAccountId, status: LikeJobStatus.RUNNING },
    });
    const otherLike = runningLike.filter((item) => item.id !== likeJobId);

    const runningParsing = await this.parsingJobs.find({
      where: { igAccountId, status: ParsingJobStatus.RUNNING },
    });

    if (otherLike.length > 0 || runningParsing.length > 0) {
      this.lLike(likeJobId, 'warn', 'job.skip', {
        reason: 'ig_account_busy',
        igAccountId,
      });
      return false;
    }

    await this.redis.releaseLock(lockKey);
    if (await this.redis.acquireLock(lockKey, ttlMs)) {
      this.lLike(likeJobId, 'debug', 'lock.acquired', { lockKey, afterStaleRelease: true });
      return true;
    }

    this.lLike(likeJobId, 'warn', 'job.skip', { reason: 'ig_account_busy', igAccountId });
    return false;
  }

  private async revertLikeOnLockFailure(job: LikeJob): Promise<void> {
    job.status = LikeJobStatus.PAUSED;
    job.errorMessage = 'IG аккаунт занят. Если другой job не идёт — нажми «Продолжить» ещё раз.';
    await this.likeJobs.save(job);
    this.events.emitLike(job.id, 'like.failed', {
      likeJobId: job.id,
      code: 'ig_account_busy',
      message: job.errorMessage,
    });
  }

  private async isLikeRunning(likeJobId: string): Promise<boolean> {
    const job = await this.likeJobs.findOne({ where: { id: likeJobId } });
    return job?.status === LikeJobStatus.RUNNING;
  }

  private async finishLikePaused(likeJobId: string): Promise<void> {
    const job = await this.likeJobs.findOne({ where: { id: likeJobId } });
    if (!job) return;
    if (job.status !== LikeJobStatus.PAUSED) {
      job.status = LikeJobStatus.PAUSED;
      await this.likeJobs.save(job);
    }
    this.lLike(likeJobId, 'info', 'job.paused', {
      likesDone: job.likesDone,
      likesSkipped: job.likesSkipped,
    });
    this.events.emitLike(likeJobId, 'like.paused', {
      likeJobId,
      likesDone: job.likesDone,
      likesSkipped: job.likesSkipped,
    });
  }

  private lLike(
    likeJobId: string,
    level: LikeLogLevel,
    event: string,
    data?: Record<string, unknown>,
  ): void {
    this.likeLog.log(likeJobId, level, event, data);
    this.events.emitLike(likeJobId, 'like.log', {
      likeJobId,
      level,
      event,
      ts: new Date().toISOString(),
      ...data,
    });
  }

  private async handleIgPostPublish(raw: string): Promise<void> {
    const { igPostId } = JSON.parse(raw) as { igPostId: string };
    const post = await this.igPosts.findOne({
      where: { id: igPostId },
      relations: { igAccount: true, mediaItems: true },
    });
    if (!post || !post.igAccount?.storageState) {
      void this.pPost(igPostId, 'warn', 'post.skip', { reason: 'missing_post_or_session' });
      return;
    }
    if (post.status !== IgPostStatus.PUBLISHING) {
      void this.pPost(igPostId, 'warn', 'post.skip', { reason: 'not_publishing', status: post.status });
      return;
    }

    const lockKey = `lock:ig:${post.igAccountId}`;
    const locked = await this.redis.acquireLock(lockKey, 15 * 60 * 1000);
    if (!locked) {
      post.status = IgPostStatus.FAILED;
      post.errorMessage = 'IG аккаунт занят другой операцией';
      await this.igPosts.save(post);
      void this.pPost(igPostId, 'error', 'post.failed', { message: post.errorMessage });
      return;
    }

    try {
      void this.pPost(igPostId, 'info', 'post.started', {
        igAccount: post.igAccount.username,
        captionLength: post.caption.length,
      });

      const imagePaths = [...(post.mediaItems ?? [])]
        .sort((a, b) => a.sortOrder - b.sortOrder)
        .map((item) => item.mediaPath);

      const result = await this.scraper.publishPhotoPost(
        post.igAccount.storageState,
        {
          imagePaths: imagePaths.length > 0 ? imagePaths : [post.mediaPath],
          caption: post.caption,
          aspectRatio: post.aspectRatio,
        },
        {
          onLog: (event, data) => {
            const level: IgPostLogLevel = event.includes('error') || event.includes('failed') ? 'error' : 'info';
            void this.pPost(igPostId, level, event, data);
          },
        },
      );

      post.status = IgPostStatus.PUBLISHED;
      post.publishedAt = new Date();
      post.igMediaId = result.igMediaId;
      post.igShortcode = result.igShortcode;
      post.errorMessage = null;
      await this.igPosts.save(post);
      void this.pPost(igPostId, 'info', 'post.completed', {
        igMediaId: result.igMediaId,
        igShortcode: result.igShortcode,
      });
    } catch (error) {
      const message = (error as Error).message;
      post.status = IgPostStatus.FAILED;
      post.errorMessage = message;
      await this.igPosts.save(post);
      void this.pPost(igPostId, 'error', 'post.failed', { message, stack: (error as Error).stack });
    } finally {
      await this.redis.releaseLock(lockKey);
    }
  }

  private pPost(
    igPostId: string,
    level: IgPostLogLevel,
    event: string,
    data?: Record<string, unknown>,
  ): void {
    void this.postLog.append(igPostId, event, data, level);
  }

  private async handleFilter(raw: string): Promise<void> {
    const { filterRunId } = JSON.parse(raw) as { filterRunId: string };
    const run = await this.filterRuns.findOne({ where: { id: filterRunId } });
    if (!run) return;

    run.status = FilterRunStatus.RUNNING;
    await this.filterRuns.save(run);
    this.events.emitFilter(filterRunId, 'filter.started', { filterRunId });

    try {
      await this.filteredResults.delete({ filterRunId });
      const profiles = await this.collectedProfiles.find({
        where: { parsingJobId: run.parsingJobId, organizationId: run.organizationId },
      });

      let matched = 0;
      for (const profile of profiles) {
        if (
          !collectedProfileMatchesFilter(profile, {
            requirePhone: run.requirePhone,
            requireNailMaster: run.requireNailMaster,
            keywords: run.keywords,
          })
        ) {
          continue;
        }

        await this.filteredResults.save(
          this.filteredResults.create({
            filterRunId: run.id,
            collectedProfileId: profile.id,
            username: profile.username,
            displayName: profile.displayName,
            bio: profile.bio,
            phone: profile.phone,
          }),
        );
        matched += 1;
        this.events.emitFilter(filterRunId, 'filter.progress', {
          filterRunId,
          processed: profiles.indexOf(profile) + 1,
          matched,
        });
      }

      run.matchedCount = matched;
      run.status = FilterRunStatus.COMPLETED;
      await this.filterRuns.save(run);
      this.events.emitFilter(filterRunId, 'filter.completed', {
        filterRunId,
        matchedCount: matched,
      });
    } catch (error) {
      run.status = FilterRunStatus.FAILED;
      await this.filterRuns.save(run);
      this.events.emitFilter(filterRunId, 'filter.failed', {
        filterRunId,
        message: (error as Error).message,
      });
    }
  }

  private async handleEnrich(raw: string): Promise<void> {
    const { enrichRunId } = JSON.parse(raw) as { enrichRunId: string };
    const run = await this.enrichRuns.findOne({
      where: { id: enrichRunId },
      relations: { igAccount: true },
    });
    if (!run || !run.igAccount?.storageState) return;

    run.status = EnrichRunStatus.RUNNING;
    await this.enrichRuns.save(run);
    this.events.emitEnrich(enrichRunId, 'enrich.started', { enrichRunId });

    const lockKey = `lock:ig:${run.igAccountId}`;
    const locked = await this.redis.acquireLock(lockKey, 60 * 60 * 1000);
    if (!locked) {
      run.status = EnrichRunStatus.FAILED;
      run.errorMessage = 'IG account is busy';
      await this.enrichRuns.save(run);
      this.events.emitEnrich(enrichRunId, 'enrich.failed', {
        enrichRunId,
        message: run.errorMessage,
      });
      return;
    }

    try {
      const results = await this.filteredResults.find({
        where: { filterRunId: run.filterRunId },
        order: { createdAt: 'ASC' },
      });
      const usernames = results.map((r) => r.username);
      const profileByUsername = new Map(results.map((r) => [r.username, r.collectedProfileId]));

      const storagePath = this.config.get<string>('storage.path', '../storage');
      let enrichedCount = 0;
      let errorsCount = 0;

      const enrichResult = await this.scraper.executeProfileEnrichment(
        run.igAccount.storageState,
        usernames,
        {
          collectAvatar: run.collectAvatar,
          recentPostsCount: run.recentPostsCount,
          storagePath,
        },
        {
          shouldContinue: async () => {
            const fresh = await this.enrichRuns.findOne({ where: { id: enrichRunId } });
            return fresh?.status === EnrichRunStatus.RUNNING;
          },
          onLog: () => undefined,
          onProfileStart: (username) => {
            this.events.emitEnrich(enrichRunId, 'enrich.progress', {
              enrichRunId,
              currentUsername: username,
              enriched: enrichedCount,
              total: usernames.length,
              errors: errorsCount,
            });
          },
          onProfileSuccess: async (media) => {
            const profileId = profileByUsername.get(media.username);
            if (!profileId) return;

            const profile = await this.collectedProfiles.findOne({ where: { id: profileId } });
            if (!profile) return;

            profile.avatarHash = media.avatarHash;
            profile.avatarPath = media.avatarPath;
            await this.collectedProfiles.save(profile);
            await this.profilesService.syncMedia(profile.id, media.recentPosts);

            if (run.audienceId) {
              await this.audiencesService.upsertFromParsing(
                run.organizationId,
                run.audienceId,
                profile,
                media.recentPosts,
              );
            }

            enrichedCount += 1;
            run.enrichedCount = enrichedCount;
            await this.enrichRuns.save(run);
            this.events.emitEnrich(enrichRunId, 'enrich.progress', {
              enrichRunId,
              currentUsername: media.username,
              enriched: enrichedCount,
              total: usernames.length,
              errors: errorsCount,
            });
          },
          onProfileError: () => {
            errorsCount += 1;
            run.errorsCount = errorsCount;
            void this.enrichRuns.save(run);
          },
          sleep: (ms) => this.scraper.sleep(ms),
          randomDelay: () => this.scraper.randomDelay(),
        },
      );

      run.enrichedCount = enrichResult.enriched;
      run.errorsCount = enrichResult.errors;
      run.status = EnrichRunStatus.COMPLETED;
      run.completedAt = new Date();
      await this.enrichRuns.save(run);
      this.events.emitEnrich(enrichRunId, 'enrich.completed', {
        enrichRunId,
        enrichedCount: enrichResult.enriched,
        errorsCount: enrichResult.errors,
      });
    } catch (error) {
      run.status = EnrichRunStatus.FAILED;
      run.errorMessage = (error as Error).message;
      await this.enrichRuns.save(run);
      this.events.emitEnrich(enrichRunId, 'enrich.failed', {
        enrichRunId,
        message: run.errorMessage,
      });
    } finally {
      await this.redis.releaseLock(lockKey);
    }
  }

  private async handleVatagoSync(raw: string): Promise<void> {
    const { vatagoSyncRunId } = JSON.parse(raw) as { vatagoSyncRunId: string };
    const run = await this.vatagoSyncRuns.findOne({ where: { id: vatagoSyncRunId } });
    if (!run) return;

    run.status = VatagoSyncRunStatus.RUNNING;
    await this.vatagoSyncRuns.save(run);
    this.events.emitVatagoSync(vatagoSyncRunId, 'vatago.started', { vatagoSyncRunId });

    const storagePath = this.config.get<string>('storage.path', '../storage');
    let exportedCount = 0;
    let skippedCount = 0;
    let errorsCount = 0;

    try {
      const rows = await this.vatagoSyncService.loadMembersForExport(run.audienceId);

      for (const row of rows) {
        const fresh = await this.vatagoSyncRuns.findOne({ where: { id: vatagoSyncRunId } });
        if (!fresh || fresh.status !== VatagoSyncRunStatus.RUNNING) break;

        const { member, avatarHash, postHashes, contacts } = row;
        this.events.emitVatagoSync(vatagoSyncRunId, 'vatago.progress', {
          vatagoSyncRunId,
          currentUsername: member.username,
          exported: exportedCount,
          skipped: skippedCount,
          errors: errorsCount,
          total: rows.length,
        });

        const phone = normalizePhoneForVatago(member.phone);
        if (!phone) {
          if (run.skipWithoutPhone) {
            skippedCount += 1;
            await this.saveVatagoResult(run.id, member, VatagoSyncMemberStatus.SKIPPED, {
              errorMessage: 'No phone',
            });
            continue;
          }
          errorsCount += 1;
          await this.saveVatagoResult(run.id, member, VatagoSyncMemberStatus.ERROR, {
            errorMessage: 'No valid phone',
          });
          continue;
        }

        try {
          const externalId = this.vatagoClient.buildExternalId(member.username);
          const importSource = this.vatagoClient.importSource;
          const status = await this.vatagoClient.getImportStatus({
            externalId,
            importSource,
            phone,
          });

          const payloadInput = {
            username: member.username,
            displayName: member.displayName,
            bio: member.bio,
            phone,
            avatarHash: status.leadInCrm && !status.needsProfileUpdate ? null : avatarHash,
            postHashes: status.leadInCrm && !status.needsProfileUpdate ? [] : postHashes,
            storagePath,
            contacts,
          };

          let response;
          let resultNote: string | undefined;
          if (status.leadInCrm && status.accountExists) {
            const { payload, files } = this.vatagoClient.buildUpdatePayload(payloadInput);
            response = await this.vatagoClient.updateMaster(payload, files);
            resultNote = 'Updated';
          } else {
            const password = this.vatagoClient.generatePassword();
            const { payload, files } = this.vatagoClient.buildPayload({
              ...payloadInput,
              password,
            });
            response = await this.vatagoClient.importMaster(payload, files);
            resultNote = response.updated ? 'Reconciled' : undefined;
          }

          exportedCount += 1;
          const memberStatus = response.updated
            ? VatagoSyncMemberStatus.SUCCESS
            : response.idempotent
              ? VatagoSyncMemberStatus.IDEMPOTENT
              : VatagoSyncMemberStatus.SUCCESS;
          await this.saveVatagoResult(run.id, member, memberStatus, {
            vatagoLeadId: response.leadId,
            vatagoProjectId: response.projectId,
            previewUrl: response.previewUrl,
            errorMessage: resultNote,
          });
        } catch (error) {
          const message = (error as Error).message;
          errorsCount += 1;
          await this.saveVatagoResult(run.id, member, VatagoSyncMemberStatus.ERROR, {
            errorMessage: message,
          });
        }

        run.exportedCount = exportedCount;
        run.skippedCount = skippedCount;
        run.errorsCount = errorsCount;
        await this.vatagoSyncRuns.save(run);

        if (run.memberDelayMs > 0) {
          await this.scraper.sleep(run.memberDelayMs);
        }
      }

      run.exportedCount = exportedCount;
      run.skippedCount = skippedCount;
      run.errorsCount = errorsCount;
      run.status = VatagoSyncRunStatus.COMPLETED;
      run.completedAt = new Date();
      await this.vatagoSyncRuns.save(run);
      this.events.emitVatagoSync(vatagoSyncRunId, 'vatago.completed', {
        vatagoSyncRunId,
        exported: exportedCount,
        skipped: skippedCount,
        errors: errorsCount,
      });
    } catch (error) {
      run.status = VatagoSyncRunStatus.FAILED;
      run.errorMessage = (error as Error).message;
      run.exportedCount = exportedCount;
      run.skippedCount = skippedCount;
      run.errorsCount = errorsCount;
      await this.vatagoSyncRuns.save(run);
      this.events.emitVatagoSync(vatagoSyncRunId, 'vatago.failed', {
        vatagoSyncRunId,
        message: run.errorMessage,
      });
    }
  }

  private async saveVatagoResult(
    runId: string,
    member: { id: string; username: string },
    status: VatagoSyncMemberStatus,
    extra: {
      vatagoLeadId?: string;
      vatagoProjectId?: string;
      previewUrl?: string;
      errorMessage?: string;
    },
  ): Promise<void> {
    await this.vatagoSyncResults.save(
      this.vatagoSyncResults.create({
        vatagoSyncRunId: runId,
        audienceMemberId: member.id,
        username: member.username,
        status,
        vatagoLeadId: extra.vatagoLeadId ?? null,
        vatagoProjectId: extra.vatagoProjectId ?? null,
        previewUrl: extra.previewUrl ?? null,
        errorMessage: extra.errorMessage ?? null,
      }),
    );
  }
}