meteora-copy/src/worker/executor.ts

import { Connection, Keypair, PublicKey } from '@solana/web3.js'
import BN from 'bn.js'

import type { ParsedDlmmOperation, WorkerSettings } from '../lib/meteora/types'
import {
	copyClosePosition,
	copyOpenPosition,
	copyRebalanceLiquidity,
	copyRemoveLiquidity,
	getSettings,
} from '../lib/meteora/dlmm-client'
import {
	createCopyTrade,
	createFollowerPosition,
	closeFollowerPosition,
	getLeaderPositionByAddress,
	updateCopyTradeStatus,
	logActivity,
} from '../lib/db/queries'
import { PositionTracker } from './position-tracker'

export class CopyExecutor {
	private settings: WorkerSettings

	constructor(
		private connection: Connection,
		private follower: Keypair,
		private tracker: PositionTracker
	) {
		this.settings = getSettings()
	}

	reloadSettings(): void {
		this.settings = getSettings()
	}

	async executeCopy(
		op: ParsedDlmmOperation,
		walletAddress: string,
		txSignature: string
	): Promise<void> {
		if (!this.settings.autoCopyEnabled) return

		try {
			switch (op.action) {
				case 'ADD_LIQUIDITY':
					await this.handleAddLiquidity(op, walletAddress, txSignature)
					break

				case 'REMOVE_LIQUIDITY':
					await this.handleRemoveLiquidity(op, walletAddress, txSignature)
					break

				case 'CLOSE_POSITION':
					await this.handleClosePosition(op, walletAddress, txSignature)
					break

				case 'REBALANCE_LIQUIDITY':
					await this.handleRebalanceLiquidity(op, walletAddress, txSignature)
					break

				case 'INIT_POSITION':
					// Only track, don't copy init alone (wait for add_liquidity)
					break
			}
		} catch (error) {
			console.error(`[Executor] Copy execution failed:`, error)
			await logActivity(
				'COPY_ERROR',
				`Copy execution failed for ${op.action}`,
				{
					error: String(error),
					walletAddress,
					txSignature,
					action: op.action,
				}
			)
		}
	}

	private async handleAddLiquidity(
		op: ParsedDlmmOperation,
		walletAddress: string,
		txSignature: string
	): Promise<void> {
		const positionAddr = op.positionAddress.toBase58()
		const lbPairAddr = op.lbPairAddress.toBase58()

		// Check if we already have a follower position for this leader position
		const existing =
			await this.tracker.getFollowerPositionForLeader(positionAddr)
		if (existing) {
			console.log(
				`[Executor] Already following position ${positionAddr}, skipping`
			)
			return
		}

		// Lookup leader position DB record
		const leaderPos = await getLeaderPositionByAddress(positionAddr)

		// Create copy trade record
		const copyTrade = await createCopyTrade({
			walletId: leaderPos?.walletId ?? '',
			leaderPositionId: leaderPos?.id,
			action: 'ADD_LIQUIDITY',
			leaderTxSignature: txSignature,
			lbPairAddress: lbPairAddr,
			leaderAmountX: op.totalXAmount?.toString(),
			leaderAmountY: op.totalYAmount?.toString(),
			leaderMinBinId: op.minBinId,
			leaderMaxBinId: op.maxBinId,
		})

		console.log(`[Executor] Copying open position on ${lbPairAddr}...`)

		try {
			const { positionAddress, txSignature: followerTx } =
				await copyOpenPosition(
					this.connection,
					this.follower,
					op,
					this.settings
				)

			// Create follower position record
			if (leaderPos) {
				await createFollowerPosition({
					leaderPositionId: leaderPos.id,
					positionAddress: positionAddress.toBase58(),
					lbPairAddress: lbPairAddr,
					lowerBinId: op.minBinId ?? 0,
					upperBinId: op.maxBinId ?? 0,
					amountXDeposited: op.totalXAmount?.toString() ?? '0',
					amountYDeposited: op.totalYAmount?.toString() ?? '0',
				})
			}

			await updateCopyTradeStatus(copyTrade.id, 'SUCCESS', {
				followerTxSignature: followerTx,
				executedAt: new Date(),
			})

			await logActivity(
				'COPY_SUCCESS',
				`Copied open position: ${positionAddress.toBase58()}`,
				{
					leaderPosition: positionAddr,
					followerPosition: positionAddress.toBase58(),
					followerTx,
				}
			)

			console.log(`[Executor] Successfully copied position: ${followerTx}`)
		} catch (error) {
			await updateCopyTradeStatus(copyTrade.id, 'FAILED', {
				errorMessage: String(error),
			})
			throw error
		}
	}

	private async handleRemoveLiquidity(
		op: ParsedDlmmOperation,
		walletAddress: string,
		txSignature: string
	): Promise<void> {
		const positionAddr = op.positionAddress.toBase58()

		const followerInfo =
			await this.tracker.getFollowerPositionForLeader(positionAddr)
		if (!followerInfo) {
			console.log(
				`[Executor] No follower position for ${positionAddr}, skipping remove`
			)
			return
		}

		const leaderPos = await getLeaderPositionByAddress(positionAddr)
		const copyTrade = await createCopyTrade({
			walletId: leaderPos?.walletId ?? '',
			leaderPositionId: leaderPos?.id,
			action: 'REMOVE_LIQUIDITY',
			leaderTxSignature: txSignature,
			lbPairAddress: followerInfo.lbPairAddress,
			leaderBpsToRemove: op.bpsToRemove,
		})

		console.log(
			`[Executor] Copying remove liquidity from ${followerInfo.positionAddress}...`
		)

		try {
			const followerTx = await copyRemoveLiquidity(
				this.connection,
				this.follower,
				new PublicKey(followerInfo.positionAddress),
				new PublicKey(followerInfo.lbPairAddress),
				op.bpsToRemove ?? 10000,
				this.settings
			)

			await updateCopyTradeStatus(copyTrade.id, 'SUCCESS', {
				followerTxSignature: followerTx,
				executedAt: new Date(),
			})

			console.log(`[Executor] Successfully removed liquidity: ${followerTx}`)
		} catch (error) {
			await updateCopyTradeStatus(copyTrade.id, 'FAILED', {
				errorMessage: String(error),
			})
			throw error
		}
	}

	private async handleRebalanceLiquidity(
		op: ParsedDlmmOperation,
		walletAddress: string,
		txSignature: string
	): Promise<void> {
		const positionAddr = op.positionAddress.toBase58()

		const followerInfo =
			await this.tracker.getFollowerPositionForLeader(positionAddr)
		if (!followerInfo) {
			console.log(
				`[Executor] No follower position for ${positionAddr}, skipping rebalance`
			)
			return
		}

		const leaderPos = await getLeaderPositionByAddress(positionAddr)
		const copyTrade = await createCopyTrade({
			walletId: leaderPos?.walletId ?? '',
			leaderPositionId: leaderPos?.id,
			action: 'REBALANCE_LIQUIDITY',
			leaderTxSignature: txSignature,
			lbPairAddress: followerInfo.lbPairAddress,
			leaderMinBinId: op.minBinId,
			leaderMaxBinId: op.maxBinId,
		})

		console.log(
			`[Executor] Copying rebalance for ${followerInfo.positionAddress}...`
		)

		try {
			const followerTx = await copyRebalanceLiquidity(
				this.connection,
				this.follower,
				new PublicKey(followerInfo.positionAddress),
				new PublicKey(followerInfo.lbPairAddress),
				op,
				this.settings
			)

			await updateCopyTradeStatus(copyTrade.id, 'SUCCESS', {
				followerTxSignature: followerTx,
				executedAt: new Date(),
			})

			await logActivity(
				'COPY_REBALANCE_SUCCESS',
				`Rebalanced follower position: ${followerInfo.positionAddress}`,
				{
					leaderPosition: positionAddr,
					followerPosition: followerInfo.positionAddress,
					followerTx,
					newMinBin: op.minBinId,
					newMaxBin: op.maxBinId,
				}
			)

			console.log(`[Executor] Successfully rebalanced position: ${followerTx}`)
		} catch (error) {
			await updateCopyTradeStatus(copyTrade.id, 'FAILED', {
				errorMessage: String(error),
			})
			throw error
		}
	}

	private async handleClosePosition(
		op: ParsedDlmmOperation,
		walletAddress: string,
		txSignature: string
	): Promise<void> {
		const positionAddr = op.positionAddress.toBase58()

		const followerInfo =
			await this.tracker.getFollowerPositionForLeader(positionAddr)
		if (!followerInfo) {
			console.log(
				`[Executor] No follower position for ${positionAddr}, skipping close`
			)
			return
		}

		const leaderPos = await getLeaderPositionByAddress(positionAddr)
		const copyTrade = await createCopyTrade({
			walletId: leaderPos?.walletId ?? '',
			leaderPositionId: leaderPos?.id,
			action: 'CLOSE_POSITION',
			leaderTxSignature: txSignature,
			lbPairAddress: followerInfo.lbPairAddress,
		})

		console.log(
			`[Executor] Copying close position ${followerInfo.positionAddress}...`
		)

		try {
			// Remove all liquidity with shouldClaimAndClose=true.
			// This removes liquidity, claims fees, AND closes the position account
			// in one step — no separate closePosition call needed.
			let followerTx: string
			try {
				followerTx = await copyRemoveLiquidity(
					this.connection,
					this.follower,
					new PublicKey(followerInfo.positionAddress),
					new PublicKey(followerInfo.lbPairAddress),
					10000,
					this.settings
				)
			} catch (removeErr) {
				// If remove fails (e.g. position already empty), try direct close
				console.log(
					`[Executor] Remove liquidity failed (${removeErr}), trying direct close...`
				)
				followerTx = await copyClosePosition(
					this.connection,
					this.follower,
					new PublicKey(followerInfo.positionAddress),
					new PublicKey(followerInfo.lbPairAddress)
				)
			}

			await closeFollowerPosition(followerInfo.positionAddress)
			await updateCopyTradeStatus(copyTrade.id, 'SUCCESS', {
				followerTxSignature: followerTx,
				executedAt: new Date(),
			})

			console.log(`[Executor] Successfully closed position: ${followerTx}`)
		} catch (error) {
			await updateCopyTradeStatus(copyTrade.id, 'FAILED', {
				errorMessage: String(error),
			})
			throw error
		}
	}
}