File-based Integration 模式

本文描述 integrations-file-based 中 legacy file parser 的代码组织方式，主要用于仍由 Retail API 直接读取的文件型 integration。它不是当前所有 file-based vendor 的统一运行架构：已迁移 vendor 由 file-based-integration-lambda → integrations/elt → Snowflake 处理，Retail API 消费 target tables。

新增或迁移 vendor 前，先确认 IntegrationType::isMigratedToElt()、Lambda vendor module 和 ELT declaration 的状态。不要为已迁移 vendor 继续新增 LATEST 文件读取或 Retail API parser 逻辑。

与 API-based 的核心区别

维度	API-based	File-based
数据源	实时 API 调用	Vendor 文件（本文只覆盖 legacy Retail API parser）
Authentication	OAuth/API Key/JWT	无（仅文件校验）
Sync Trigger	手动 + 定时 + Webhook	Legacy 定时/手动；ELT vendor 由 SQS/SNS pipeline 驱动
State Management	复杂（token refresh、重试）	简单（解析成功或失败）
Error Recovery	自动重试	用户重新上传
Data Freshness	实时或接近实时	取决于文件上传时间

共享 Package

位置：/packages/libs/integrations-file-based/

Legacy file-based integrations 使用该 package 提供以下通用能力：

文件上传处理
格式校验
解析工具
错误处理

类结构模式

app/Integrations/[Vendor]/
├── Integrator.php       # Extends file-based base, main entry point
├── Parser.php           # File format parsing (CSV/Excel)
├── Mapper.php           # Map parsed data to internal models
├── Validator.php        # File and data validation
├── Models/              # Vendor-specific data structures
│   ├── Account.php
│   ├── Holding.php
│   └── Position.php
└── Exceptions/          # Vendor-specific errors

处理流程

┌─────────────────────────────────────────────────────────────────────────────┐
│                           User Uploads File                                  │
│                    (via UI or API endpoint)                                  │
└─────────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                          File Storage (S3)                                   │
│                                                                              │
│   Path: integrations/{integration_id}/{reference}/file.csv                   │
│   Reference from: $integration->file_reference                               │
└─────────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                         1. File Validation                                   │
│                                                                              │
│   - Check file exists                                                        │
│   - Validate file size (not too large)                                       │
│   - Check file format (CSV/Excel)                                           │
│   - Validate encoding (UTF-8, Latin-1, etc.)                                │
│   - Check file freshness (not outdated)                                     │
└─────────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                         2. Parse File                                        │
│                                                                              │
│   Parser::parse($filePath)                                                   │
│   - Read file contents                                                       │
│   - Parse CSV rows or Excel sheets                                          │
│   - Extract headers and data rows                                            │
│   - Handle format-specific quirks                                           │
└─────────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                         3. Map to Vendor Models                              │
│                                                                              │
│   Mapper::map($parsedData)                                                   │
│   - Map columns to vendor model fields                                       │
│   - Transform data types (strings to numbers, dates)                        │
│   - Handle null/empty values                                                │
│   - Create vendor-specific Account/Holding objects                          │
└─────────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                         4. Map to Internal Models                            │
│                                                                              │
│   Integrator::sync($mapping)                                                 │
│   - Match vendor accounts to existing IntegrationMappings                   │
│   - Create/Update Account, Insurance, Holding records                        │
│   - Update IntegrationMapping.last_completed_at                             │
└─────────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                         5. Return Results                                    │
│                                                                              │
│   - Imported households                                                      │
│   - Created/updated accounts                                                │
│   - Error messages for failed rows                                          │
└─────────────────────────────────────────────────────────────────────────────┘

文件存储

S3 路径结构

// Integration model method
public function getFileReferenceAttribute(): string
{
    return $this->type->getFileReference($this->reference);
}

// IntegrationType enum method
public function getFileReference(string $reference): string
{
    // Returns S3 path prefix for this integration's files
    return "integrations/{$this->value}/{$reference}/";
}

Host 类型

File-based integrations 有两种 host 类型：

// IntegrationType enum
public function getFileBasedHostType(): string
{
    return match($this) {
        self::FIDELITY, self::PERSHING => 'self-hosted',    // Vendor pushes to our FTP
        self::APEX, self::LPL => 'vendor-hosted',           // We pull from vendor
        // ...
    };
}

Host Type	说明	流程
`self-hosted`	数据存储在 RC 服务器上（vendor 推送到 RC FTP）	Vendor → RC FTP
`vendor-hosted`	数据存储在 vendor 服务器上（RC 从 vendor 拉取）	RC → Vendor

Parser 设计

Parser 基础接口

interface Parser
{
    /**
     * Parse file and return structured data
     *
     * @param string $filePath Path to uploaded file
     * @return array<int, array<string, mixed>> Parsed rows
     * @throws FileNotFoundException
     * @throws InvalidFormatException
     */
    public function parse(string $filePath): array;
}

CSV Parser 示例

class CsvParser implements Parser
{
    public function __construct(
        private string $delimiter = ',',
        private string $enclosure = '"',
        private string $encoding = 'UTF-8'
    ) {}

    public function parse(string $filePath): array
    {
        $content = file_get_contents($filePath);

        // Handle encoding
        if ($this->encoding !== 'UTF-8') {
            $content = mb_convert_encoding($content, 'UTF-8', $this->encoding);
        }

        $lines = str_getcsv($content, "\n");
        $headers = str_getcsv(array_shift($lines), $this->delimiter, $this->enclosure);

        $rows = [];
        foreach ($lines as $line) {
            if (empty(trim($line))) continue;

            $values = str_getcsv($line, $this->delimiter, $this->enclosure);
            $rows[] = array_combine($headers, $values);
        }

        return $rows;
    }
}

Excel Parser 示例

class ExcelParser implements Parser
{
    public function __construct(
        private ?string $sheetName = null,
        private int $headerRow = 1
    ) {}

    public function parse(string $filePath): array
    {
        $spreadsheet = IOFactory::load($filePath);

        $sheet = $this->sheetName
            ? $spreadsheet->getSheetByName($this->sheetName)
            : $spreadsheet->getActiveSheet();

        $data = $sheet->toArray();

        // Extract headers from specified row
        $headers = $data[$this->headerRow - 1];

        // Map remaining rows
        $rows = [];
        for ($i = $this->headerRow; $i < count($data); $i++) {
            if (empty(array_filter($data[$i]))) continue;
            $rows[] = array_combine($headers, $data[$i]);
        }

        return $rows;
    }
}

Mapper 设计

Mapper 基础接口

interface Mapper
{
    /**
     * Map parsed row to vendor model
     *
     * @param array<string, mixed> $row Parsed row data
     * @return Account|Holding|null Mapped model or null if invalid
     */
    public function map(array $row): Account|Holding|null;

    /**
     * Get column mapping configuration
     *
     * @return array<string, string> Vendor column => model field
     */
    public function getColumnMapping(): array;
}

Mapper 示例

说明：这是 CSV integration 的通用示例。实际的 Fidelity integration 使用 self-hosted fixed-width 文件，并在 integrations-file-based package 中使用 Reader.php + Extractor.php，详见 Fidelity vendor 文档。

class FidelityAccountMapper implements Mapper
{
    public function getColumnMapping(): array
    {
        return [
            'Account Number' => 'account_number',
            'Account Name' => 'name',
            'Account Type' => 'type',
            'Total Value' => 'balance',
            'As of Date' => 'as_of_date',
        ];
    }

    public function map(array $row): ?Account
    {
        $accountNumber = $row['Account Number'] ?? null;
        if (empty($accountNumber)) {
            return null;
        }

        return new Account(
            reference: $accountNumber,
            name: $row['Account Name'] ?? 'Unknown',
            type: $this->mapAccountType($row['Account Type'] ?? ''),
            balance: $this->parseAmount($row['Total Value'] ?? '0'),
            asOfDate: $this->parseDate($row['As of Date'] ?? ''),
        );
    }

    private function parseAmount(string $value): float
    {
        // Remove currency symbols, commas, handle negatives
        $cleaned = preg_replace('/[^0-9.\-]/', '', $value);
        return (float) $cleaned;
    }

    private function parseDate(string $value): ?Carbon
    {
        // Handle MM/DD/YYYY format
        try {
            return Carbon::createFromFormat('m/d/Y', $value);
        } catch (Exception) {
            return null;
        }
    }

    private function mapAccountType(string $vendorType): string
    {
        return match(strtoupper($vendorType)) {
            'IRA', 'ROTH IRA', 'TRADITIONAL IRA' => 'retirement',
            'BROKERAGE', 'INDIVIDUAL' => 'investment',
            'JOINT', 'JTWROS' => 'investment',
            '401K', '403B' => 'retirement',
            default => 'investment',
        };
    }
}

Integrator 设计

File-based Integrator 基类

abstract class FileBasedIntegrator extends Integrator
{
    abstract protected function getParser(): Parser;
    abstract protected function getMapper(): Mapper;

    public static function sync(IntegrationMapping $mapping): ?array
    {
        $integrator = new static($mapping->integration);

        // Get file path from S3
        $filePath = $integrator->downloadFile($mapping);

        // Validate file freshness
        if ($integrator->isFileOutdated($filePath)) {
            throw new FileOutdatedException(
                $mapping->integration,
                AffectedLevel::IntegrationMapping
            );
        }

        // Parse file
        $parsedRows = $integrator->getParser()->parse($filePath);

        // Map and import
        $results = [];
        foreach ($parsedRows as $row) {
            $entity = $integrator->getMapper()->map($row);
            if ($entity) {
                $results[] = $integrator->importEntity($entity, $mapping);
            }
        }

        $mapping->last_completed_at = now();
        $mapping->save();

        return $results;
    }

    protected function isFileOutdated(string $filePath): bool
    {
        $maxAge = config('integrations.file_max_age_days', 30);
        $fileDate = Carbon::createFromTimestamp(filemtime($filePath));

        return $fileDate->diffInDays(now()) > $maxAge;
    }
}

Vendor Integrator 示例

Note: This is a generic example for CSV-based integrations. The real Fidelity integration (self-hosted fixed-width) does not use CsvParser — see Fidelity vendor doc.

class FidelityIntegrator extends FileBasedIntegrator
{
    protected function getParser(): Parser
    {
        return new CsvParser(delimiter: ',', encoding: 'UTF-8');
    }

    protected function getMapper(): Mapper
    {
        return new FidelityAccountMapper();
    }

    protected static function getVendor(): IntegrationType
    {
        return IntegrationType::FIDELITY;
    }
}

校验

文件校验

class FileValidator
{
    public function validate(string $filePath): void
    {
        // Check file exists
        if (!file_exists($filePath)) {
            throw new FileNotFoundException("File not found: {$filePath}");
        }

        // Check file size
        $maxSize = config('integrations.max_file_size', 50 * 1024 * 1024); // 50MB
        if (filesize($filePath) > $maxSize) {
            throw new FileTooLargeException("File exceeds {$maxSize} bytes");
        }

        // Validate format
        $extension = pathinfo($filePath, PATHINFO_EXTENSION);
        if (!in_array(strtolower($extension), ['csv', 'xlsx', 'xls'])) {
            throw new InvalidFormatException("Unsupported format: {$extension}");
        }
    }
}

数据校验

class DataValidator
{
    public function validateRow(array $row, array $requiredFields): array
    {
        $errors = [];

        foreach ($requiredFields as $field) {
            if (empty($row[$field])) {
                $errors[] = "Missing required field: {$field}";
            }
        }

        return $errors;
    }
}

错误处理

文件专属 Exceptions

// File not found in S3
class FileNotFoundException extends ExternalServiceException
{
    protected function getDefaultHttpResponseMessage(): string
    {
        return 'The data file was not found. Please upload a new file.';
    }
}

// File is too old
class FileOutdatedException extends ExternalServiceException
{
    protected function getDefaultHttpResponseMessage(): string
    {
        return 'The data file is outdated. Please upload a recent file.';
    }
}

// Invalid file format
class InvalidFormatException extends ExternalServiceException
{
    protected function getDefaultHttpResponseMessage(): string
    {
        return 'The file format is not supported. Please upload a CSV or Excel file.';
    }
}

行级错误处理

public static function sync(IntegrationMapping $mapping): ?array
{
    $results = [];
    $errors = [];

    foreach ($parsedRows as $index => $row) {
        try {
            $entity = $mapper->map($row);
            if ($entity) {
                $results[] = $this->importEntity($entity, $mapping);
            }
        } catch (Exception $e) {
            $errors[] = [
                'row' => $index + 2, // +2 for header row and 0-indexing
                'error' => $e->getMessage(),
                'data' => $row,
            ];
        }
    }

    if (!empty($errors)) {
        Log::warning('File import had errors', [
            'integration_id' => $mapping->integration_id,
            'error_count' => count($errors),
            'errors' => array_slice($errors, 0, 10), // Log first 10
        ]);
    }

    return ['imported' => $results, 'errors' => $errors];
}