OCS Audience Service và luồng Segment nguồn OCS

1. Mục tiêu

Tài liệu này mô tả phương án lấy tệp khách hàng từ OCS/Hadoop để sử dụng trong Segment và Campaign của LaoAds V2.

Thiết kế cần đáp ứng các yêu cầu sau:

LaoAds không kết nối trực tiếp tới Hadoop của OCS.
Một service độc lập được đóng gói thành JAR và triển khai trong môi trường OCS.
OCS Audience Service chỉ cần quyền đọc Hadoop qua JDBC của Hive, Impala hoặc Trino.
Người dùng LaoAds tạo Segment bằng cách chọn bảng, chọn cột và tạo điều kiện lọc; không nhập SQL.
Danh sách bảng và cột được phép dùng trên giao diện do LaoAds quản lý trong MongoDB.
Kết quả lớn được truyền theo dạng stream và ghi theo batch; không tải toàn bộ dữ liệu vào RAM.
Mỗi lần Campaign chạy sử dụng một SegmentVersion bất biến và một TargetSnapshot đã được chốt.
Luồng phải chạy được trên nhiều node, tránh build trùng dữ liệu và có đủ log để truy vết.

2. Phạm vi thay đổi

2.1. Phần giữ nguyên

Các thành phần sau tiếp tục được sử dụng:

Collection segment và giá trị type = QUERY.
segmentType = DYNAMIC và refreshMode = PER_RUN đối với Segment nguồn OCS.
Collection segment_version để quản lý từng lần build dữ liệu.
ClickHouse segment_version_member để lưu danh sách khách hàng lớn.
CampaignRun, TargetSnapshot và ClickHouse campaign_target_member.
Luồng FILE và EVENT không thuộc phạm vi thay đổi của tài liệu này.

2.2. Phần loại bỏ khỏi luồng Segment QUERY

Không còn nguồn LaoID khi tạo Segment QUERY.
Không còn truy vấn trực tiếp bảng user_basic_info bằng JdbcTemplate trong LaoAds.
Không sử dụng CustomerQueryBuilder hiện tại cho OCS.
Không cho phép SELECT * và không nối trực tiếp giá trị người dùng vào SQL.
Không xây dựng lớp Query Template, Query Template Version, Publish hoặc Revoke.
Không dùng GraphQL cho luồng lấy audience.

user_lao_id_field_config, UserFieldController và các lớp LaoID liên quan chỉ được xóa sau khi rà soát chắc chắn không còn module khác sử dụng. Nếu vẫn có luồng cũ phụ thuộc, đánh dấu TODO và tách việc loại bỏ sang task riêng.

Luồng OCS kế thừa cách dựng form điều kiện của user_lao_id_field_config, nhưng không dùng chung collection. OCS có nhiều bảng và cần thêm metadata về quyền select, quyền filter, operator cùng cách ánh xạ kết quả nên phải có model riêng.

2.3. Các thành phần mới

ocs_dataset_config
    Danh mục bảng OCS mà LaoAds cho phép hiển thị trên FE.

ocs_dataset_field_config
    Danh mục cột của từng bảng, phát triển từ pattern user_lao_id_field_config.

queryDefinition trong segment
    Định nghĩa bảng, cột và điều kiện lọc mà người dùng đã chọn.

segment_build_job
    Theo dõi tiến độ build SegmentVersion, retry và quyền xử lý giữa nhiều worker.

OCS Audience Service
    Kiểm tra query definition, đọc Hadoop và stream kết quả về Segment Import Worker.

3. Kiến trúc tổng thể

[Diagram]

3.1. Trách nhiệm của từng hệ thống

Hệ thống	Trách nhiệm
Frontend	Hiển thị danh sách bảng/cột do LaoAds cung cấp, tạo điều kiện lọc và gửi cấu hình Segment.
LaoAds API	Quản lý danh mục OCS, validate request, gọi OCS verify, lưu Segment và cung cấp API cho FE.
MongoDB	Lưu metadata, trạng thái SegmentVersion, CampaignRun và build job; không lưu hàng triệu member.
Segment Import Worker	Gọi stream OCS, normalize dữ liệu, insert ClickHouse theo batch và quản lý retry.
OCS Audience Service	Kiểm tra bảng/cột thực tế, tạo câu SQL an toàn, bind parameter và stream ResultSet.
Hadoop	Nguồn dữ liệu OCS, chỉ cho tài khoản service đọc.
ClickHouse	Lưu member dung lượng lớn và phục vụ xem, tạo target, thống kê.

4. Luồng giao diện

[Diagram]

4.1. API FE sử dụng để tạo Segment

GET  /ocs-datasets
GET  /ocs-datasets/{datasetCode}/columns
POST /segments
PUT  /segments/{segmentId}

Hai API /ocs-datasets đọc dữ liệu từ MongoDB của LaoAds, không gọi sang OCS theo từng request của FE.

Cách dựng giao diện giữ cùng nguyên tắc với luồng LaoID cũ:

user_lao_id_field_config cũ trả metadata field để FE dựng điều kiện.
Luồng OCS trả ocs_dataset_field_config của bảng đã chọn.
FE dựa vào dataType để chọn loại input và dựa vào allowedOperators để hiển thị toán tử.
FE gửi fieldCode; backend mới resolve sang columnName vật lý.

Khác biệt là LaoID cũ chỉ có một bảng user_basic_info, còn OCS có nhiều bảng. Vì vậy FE phải chọn datasetCode trước rồi mới tải field của dataset đó.

Response field cho FE tham khảo:

[
  {
    "fieldCode": "DATA_USAGE_MB",
    "label": "Dung lượng Data đã sử dụng",
    "dataType": "NUMBER",
    "selectable": true,
    "filterable": true,
    "targetField": "ATTRIBUTE",
    "allowedOperators": [
      "EQUALS",
      "GREATER_THAN",
      "GREATER_THAN_OR_EQUALS",
      "LESS_THAN",
      "BETWEEN"
    ]
  }
]

Backend chọn label theo X-Accept-Language giống API /user-field hiện tại; FE không phải tự chọn giữa labelVi, labelLo, labelEn và labelZh.

4.2. Các trường FE cần gửi

Request tham khảo:

{
  "name": "Thuê bao có nhu cầu nâng cấp gói Data",
  "description": "Tệp chạy chiến dịch nâng cấp Data",
  "type": "QUERY",
  "segmentType": "DYNAMIC",
  "queryDefinition": {
    "datasetCode": "SUBSCRIBER_DAILY",
    "selectedFields": [
      "MSISDN",
      "FIRST_NAME",
      "PACKAGE_CODE"
    ],
    "conditions": [
      {
        "fieldCode": "SNAPSHOT_DATE",
        "operator": "EQUALS",
        "valueSource": "RUN_DATE",
        "dayOffset": -1
      },
      {
        "fieldCode": "DATA_USAGE_MB",
        "operator": "GREATER_THAN_OR_EQUALS",
        "valueSource": "FIXED",
        "value": 5120,
        "relation": "AND"
      }
    ]
  }
}

Quy tắc:

FE chỉ gửi datasetCode và fieldCode, không gửi tên bảng/cột vật lý tùy ý.
Backend lấy schemaName, tableName từ ocs_dataset_config và lấy columnName từ ocs_dataset_field_config.
selectedFields phải chứa ít nhất một trường được ánh xạ thành PHONE hoặc EMAIL.
operator phải thuộc danh sách allowedOperators của field.
valueSource = FIXED dùng giá trị cố định do người dùng nhập.
valueSource = RUN_DATE được resolve theo ngày chạy thực tế của CampaignRun.
FE không được gửi raw SQL.

4.3. Màn chi tiết Segment

Màn chi tiết hiển thị:

Bảng OCS đang dùng.
Các cột đã chọn.
Điều kiện lọc.
Thời điểm cấu hình được verify gần nhất.
Danh sách SegmentVersion đã build.
Mặc định hiển thị member của latestReadyVersionId; người dùng có thể chọn version cũ để đối soát.

Khi Segment mới được tạo nhưng chưa chạy Campaign, danh sách version có thể rỗng. Đây là trạng thái hợp lệ vì Segment QUERY chỉ lưu định nghĩa; dữ liệu được build khi Campaign chạy.

5. Danh mục bảng và cột OCS trong MongoDB

5.1. Collection `ocs_dataset_config`

Mỗi document đại diện cho một bảng OCS mà LaoAds cho phép người dùng chọn.

{
  "_id": "subscriber-daily",
  "datasetCode": "SUBSCRIBER_DAILY",
  "displayName": "Thông tin thuê bao theo ngày",
  "description": "Dữ liệu thuê bao được chốt theo ngày",
  "schemaName": "ocs",
  "tableName": "subscriber_daily",
  "status": "ACTIVE",
  "requiredFilterFieldCodes": [
    "SNAPSHOT_DATE"
  ],
  "maxQueryRangeDays": 31,
  "createdAt": "2026-07-29T00:00:00Z",
  "updatedAt": "2026-07-29T00:00:00Z"
}

Các field thuộc bảng không nhúng vào document này. Chúng được lưu riêng trong ocs_dataset_field_config để giữ cách quản lý gần giống user_lao_id_field_config và tránh document bảng tăng kích thước khi số cột nhiều.

5.2. Collection `ocs_dataset_field_config`

Mỗi document đại diện cho một cột được phép hiển thị hoặc dùng làm điều kiện của một dataset.

{
  "_id": "subscriber-daily-msisdn",
  "datasetCode": "SUBSCRIBER_DAILY",
  "fieldCode": "MSISDN",
  "columnName": "isdn",
  "label": "Phone number",
  "labelEn": "Phone number",
  "labelVi": "Số thuê bao",
  "labelLo": "ເບີໂທ",
  "labelZh": "电话号码",
  "dataType": "STRING",
  "selectable": true,
  "filterable": true,
  "targetField": "PHONE",
  "allowedOperators": [
    "EQUALS",
    "IN"
  ],
  "status": "ACTIVE",
  "sortOrder": 1,
  "createdAt": "2026-07-29T00:00:00Z",
  "updatedAt": "2026-07-29T00:00:00Z"
}

Ví dụ field chỉ dùng để lọc theo ngày:

{
  "datasetCode": "SUBSCRIBER_DAILY",
  "fieldCode": "SNAPSHOT_DATE",
  "columnName": "snapshot_date",
  "labelEn": "Snapshot date",
  "labelVi": "Ngày dữ liệu",
  "labelLo": "ວັນທີຂໍ້ມູນ",
  "labelZh": "数据日期",
  "dataType": "DATE",
  "selectable": false,
  "filterable": true,
  "targetField": "ATTRIBUTE",
  "allowedOperators": [
    "EQUALS",
    "BETWEEN"
  ],
  "status": "ACTIVE",
  "sortOrder": 2
}

5.3. So sánh với `user_lao_id_field_config`

`user_lao_id_field_config` cũ	`ocs_dataset_field_config` mới	Ghi chú
`key`	`fieldCode`	Mã field FE gửi về.
`table`	Quan hệ qua `datasetCode`	OCS phải chọn dataset trước vì có nhiều bảng.
`fieldName` hoặc `key`	`columnName`	Tên cột vật lý, không trả cho FE nếu không cần.
`label`, `labelEn`, `labelVi`, `labelLo`, `labelZh`	Giữ nguyên	Tiếp tục hỗ trợ nhãn theo ngôn ngữ hiện tại.
`dataType`	Giữ nguyên nhưng chuyển thành enum	FE dùng để dựng đúng input.
Chưa có	`selectable`	Xác định cột có được lấy ra làm dữ liệu người nhận không.
Chưa có	`filterable`	Xác định cột có được viết điều kiện không.
Chưa có	`allowedOperators`	Không cho FE tự gửi operator bất kỳ.
Chưa có	`targetField`	Ánh xạ về phone, email, tên hoặc attribute.
Chưa có	`status`, `sortOrder`	Ẩn field và sắp xếp giao diện mà không xóa dữ liệu.

Không tái sử dụng trực tiếp UserLaoIdFieldConfigModel vì model đó có key unique toàn hệ thống. Trong OCS, hai bảng khác nhau có thể cùng có field MSISDN, nên uniqueness phải nằm trong phạm vi datasetCode.

5.4. Ý nghĩa các trường

Trường	Ý nghĩa
`datasetCode`	Mã ổn định FE và Segment sử dụng; không phụ thuộc tên bảng hiển thị.
`schemaName`, `tableName`	Bảng vật lý trong Hadoop. Chỉ backend sử dụng.
`status`	`ACTIVE` cho phép tạo/sửa Segment; `INACTIVE` ẩn khỏi lựa chọn mới.
`requiredFilterFieldCodes`	Các điều kiện bắt buộc để tránh query quét toàn bộ Hadoop.
`maxQueryRangeDays`	Khoảng ngày tối đa được phép query nếu dataset có filter theo thời gian.
`fieldCode`	Mã field ổn định FE gửi về.
`columnName`	Tên cột vật lý trong Hadoop.
`dataType`	`STRING`, `NUMBER`, `DATE`, `DATETIME`, `BOOLEAN`.
`selectable`	Field có được đưa vào kết quả hay không.
`filterable`	Field có được dùng làm điều kiện lọc hay không.
`targetField`	Ánh xạ kết quả về `PHONE`, `EMAIL`, `FIRST_NAME`, `LAST_NAME` hoặc `ATTRIBUTE`.
`allowedOperators`	Danh sách toán tử FE được phép hiển thị và backend chấp nhận.
`labelEn`, `labelVi`, `labelLo`, `labelZh`	Nhãn field theo ngôn ngữ hiện tại của người dùng.
`sortOrder`	Thứ tự field trên giao diện.

5.5. Index MongoDB

ocs_dataset_config:
  (datasetCode) unique
  (schemaName, tableName) unique
  (status, displayName)

ocs_dataset_field_config:
  (datasetCode, fieldCode) unique
  (datasetCode, columnName) unique
  (datasetCode, status, sortOrder)

API lấy field luôn có datasetCode; các index bắt đầu bằng datasetCode để query đúng phạm vi bảng và không quét toàn collection.

5.6. Cách cập nhật khi OCS thay đổi schema

[Diagram]

Hai API catalog của OCS chỉ dành cho kỹ sư kiểm tra bằng Swagger, curl hoặc Postman. LaoAds không gọi hai API này trong luồng tạo Segment hoặc chạy Campaign.

6. Dữ liệu lưu trong `segment`

6.1. Bổ sung `queryDefinition`

{
  "type": "QUERY",
  "segmentType": "DYNAMIC",
  "refreshMode": "PER_RUN",
  "queryDefinition": {
    "datasetCode": "SUBSCRIBER_DAILY",
    "schemaName": "ocs",
    "tableName": "subscriber_daily",
    "selectedColumns": [
      {
        "fieldCode": "MSISDN",
        "columnName": "isdn",
        "dataType": "STRING",
        "targetField": "PHONE"
      }
    ],
    "conditions": [],
    "definitionChecksum": "sha256...",
    "verifiedAt": "2026-07-29T10:00:00Z"
  }
}

LaoAds lưu cả mã logic và tên vật lý đã được resolve tại thời điểm lưu Segment. Vì vậy việc sửa ocs_dataset_config hoặc ocs_dataset_field_config không âm thầm làm thay đổi một Segment đang hoạt động.

Nếu schema OCS thay đổi:

Cập nhật ocs_dataset_config và ocs_dataset_field_config.
Mở và lưu lại Segment liên quan.
Backend resolve lại column, gọi OCS verify và tạo definitionChecksum mới.
Các CampaignRun đã tạo trước đó vẫn giữ source snapshot cũ; Run mới dùng định nghĩa mới.

6.2. `SegmentSourceSnapshotModel`

Khi build một version, phải snapshot đầy đủ:

type
queryDefinition
resolvedConditions
sourceDataAsOf
definitionChecksum

resolvedConditions là điều kiện sau khi các giá trị như RUN_DATE đã được chuyển thành giá trị cụ thể. Snapshot này không được sửa sau khi SegmentVersion được tạo.

6.3. `segment_version`

Giữ collection hiện tại và bổ sung thông tin nguồn cần thiết:

companyId
segmentId
versionNo
sourceType = QUERY
sourceDefinitionSnapshot
status = BUILDING | READY | FAILED | ARCHIVED
totalMemberCount
validMemberCount
invalidMemberCount
builtAt
completedAt
errorSummary

Version READY là bất biến. Sửa Segment không cập nhật member hoặc source snapshot của version cũ.

7. Hợp đồng OCS Audience Service

7.1. API catalog dùng để kiểm tra thủ công

GET /internal/v1/catalog/tables
GET /internal/v1/catalog/tables/{schemaName}/{tableName}/columns

API chỉ trả các bảng/cột mà JDBC service account nhìn thấy. API phải được bảo vệ bằng mạng nội bộ và xác thực dành cho kỹ sư vận hành.

7.2. API verify được LaoAds gọi

POST /internal/v1/audiences/verify

Request tham khảo:

{
  "requestId": "verify-uuid",
  "schemaName": "ocs",
  "tableName": "subscriber_daily",
  "selectedColumns": [
    {
      "columnName": "isdn",
      "dataType": "STRING",
      "targetField": "PHONE"
    }
  ],
  "conditions": [
    {
      "columnName": "snapshot_date",
      "dataType": "DATE",
      "operator": "EQUALS",
      "value": "2026-07-28"
    }
  ]
}

OCS Service kiểm tra:

Schema, table và column có tồn tại theo JDBC metadata.
Kiểu dữ liệu của value phù hợp với cột.
Operator nằm trong tập toán tử service hỗ trợ.
Chỉ tạo câu SELECT; không có DDL, DML, raw expression, subquery hoặc SQL do FE truyền lên.
Câu query có thể prepare/parse thành công.
Có thể dùng EXPLAIN hoặc query giới hạn 0/1 tùy khả năng của Hive, Impala hoặc Trino; không chạy full query khi verify.

Response tham khảo:

{
  "success": true,
  "data": {
    "valid": true,
    "queryChecksum": "sha256...",
    "warnings": []
  }
}

7.3. API stream được Worker gọi

POST /internal/v1/audiences/stream
Content-Type: application/json
Accept: application/x-ndjson

Request chứa:

requestId
companyId
segmentId
segmentVersionId
importAttemptId
schemaName
tableName
selectedColumns
resolvedConditions
queryChecksum

OCS Service không tin queryChecksum thay cho validation. Service vẫn kiểm tra request và compile câu query từ cấu trúc đã nhận trước khi thực thi.

7.4. Định dạng NDJSON

Mỗi dòng là một JSON độc lập:

{"recordType":"START","requestId":"...","startedAt":"..."}
{"recordType":"DATA","phone":"85620...","email":null,"firstName":"Somchai","lastName":null,"attributes":{"packageCode":"PKG01"}}
{"recordType":"DATA","phone":"85620...","email":"user@example.com","firstName":null,"lastName":null,"attributes":{}}
{"recordType":"END","requestId":"...","rowCount":250000,"completedAt":"..."}

Nếu lỗi xảy ra trước khi response bắt đầu, OCS trả HTTP status và error body chuẩn. Nếu lỗi xảy ra sau khi stream đã bắt đầu, OCS ghi một dòng ERROR cuối cùng rồi đóng stream:

{"recordType":"ERROR","errorCode":"OCS_STREAM0004","message":"Hadoop query interrupted"}

Worker chỉ chuyển version sang READY khi nhận đúng END và đối soát được số dòng. Mất kết nối hoặc nhận ERROR luôn được coi là attempt thất bại.

7.5. Tại sao stream không làm tràn RAM

OCS Service phải:

Dùng JDBC forward-only ResultSet nếu driver hỗ trợ.
Cấu hình fetchSize phù hợp driver, ví dụ 1.000-5.000 dòng.
Chuyển từng row thành một dòng NDJSON và ghi ngay ra response.
Không gọi resultSet thành List.
Không tạo file tạm chứa toàn bộ kết quả.
Có giới hạn số query chạy đồng thời, timeout và hủy statement khi client ngắt kết nối.

Segment Import Worker phải:

Đọc response theo từng dòng.
Giữ tối đa một batch, mặc định 1.000 member, trong RAM.
Insert ClickHouse theo batch.
Xóa batch khỏi RAM ngay sau khi insert thành công.
Chỉ cập nhật tiến độ Mongo theo chu kỳ, ví dụ mỗi 10.000 dòng hoặc 5 giây; không update Mongo cho từng dòng.

8. Luồng tạo và sửa Segment

[Diagram]

Khi sửa Segment:

Chỉ cho sửa khi quyền và trạng thái Segment cho phép.
Mọi thay đổi dataset, selected field hoặc condition đều phải verify lại.
Không sửa SegmentVersion đã có.
CampaignRun đã chốt source snapshot không bị ảnh hưởng.
Run mới sử dụng định nghĩa Segment mới nhất tại thời điểm chuẩn bị target.

9. Luồng CampaignRun và build SegmentVersion

[Diagram]

9.1. `buildKey` và trường hợp nhiều Campaign dùng chung Segment

buildKey được tạo từ:

companyId
segmentId
definitionChecksum
resolvedConditions
sourceDataAsOf

Sau đó canonicalize dữ liệu và băm SHA-256.

Nếu nhiều CampaignRun cùng sử dụng đúng một Segment, cùng điều kiện đã resolve và cùng thời điểm dữ liệu:

Run đầu tiên tạo SegmentVersion và build job.
Các Run còn lại tham chiếu cùng version đang build và chờ kết quả.
Chỉ một worker được quyền query Hadoop cho buildKey đó.
Khi version READY, từng Run tạo TargetSnapshot riêng.

Nếu ngày dữ liệu, điều kiện hoặc định nghĩa Segment khác nhau thì buildKey khác và phải build version mới.

9.2. Collection `segment_build_job`

id
companyId
segmentId
segmentVersionId
buildKey
status: QUEUED | RUNNING | RETRY_WAIT | COMPLETED | FAILED
attemptNo
currentAttemptId
leaseOwner
leaseUntil
lastHeartbeatAt
processedCount
validCount
invalidCount
nextRetryAt
errorCode
errorSummary
createdAt
updatedAt

Index bắt buộc:

(buildKey) unique
(status, nextRetryAt)
(status, leaseUntil)
(segmentId, createdAt desc)

leaseOwner và leaseUntil đảm bảo nhiều worker không xử lý đồng thời cùng một job. Worker phải gia hạn lease trong khi stream; node khác chỉ được claim khi lease hết hạn.

9.3. Bổ sung ClickHouse `segment_version_member`

Giữ schema hiện tại và bổ sung:

import_attempt_id String
source_data_as_of Nullable(DateTime64(3))

import_attempt_id giúp phân biệt dữ liệu của từng lần thử. Khi retry, dữ liệu của attempt thất bại không được tính vào version hoàn chỉnh.

Phương án đọc an toàn:

segment_version lưu successfulAttemptId khi build thành công.
API member và bước tạo TargetSnapshot chỉ đọc import_attempt_id = successfulAttemptId.
Dữ liệu attempt thất bại được cleanup nền sau đó; không chặn retry.

10. Xử lý lỗi và retry

[Diagram]

Quy tắc:

Retry hữu hạn, ví dụ tối đa 3 attempt với exponential backoff.
Không tiếp tục từ dòng HTTP đã đứt vì JDBC stream không có checkpoint ổn định theo mọi engine.
Retry chạy lại toàn bộ query với importAttemptId mới.
Attempt cũ không làm trùng target vì chỉ successful attempt được đọc.
Sau số lần retry tối đa, version và build job chuyển FAILED; CampaignRun chuyển FAILED với mã lỗi rõ ràng.
Không dispatch Campaign khi SegmentVersion chưa READY hoặc TargetSnapshot chưa được chốt.

11. Phân chia task

Các task dưới đây đủ lớn để giao độc lập nhưng vẫn có hợp đồng rõ giữa các nhóm. Không tách riêng task kiểm thử; mỗi task phải có unit test, integration test và log trong phạm vi của task đó.

Task 1 - Xây dựng OCS Audience Service

Mục tiêu

Tạo JAR độc lập triển khai trong OCS, có khả năng kiểm tra metadata, verify cấu hình audience và stream dữ liệu Hadoop mà không giữ toàn bộ kết quả trong RAM.

Phạm vi

Tạo Spring Boot service độc lập, không phụ thuộc code nghiệp vụ LaoAds.
Cấu hình JDBC cho đúng engine thực tế: Hive, Impala hoặc Trino.
Dùng JDBC user read-only và giới hạn schema được phép đọc.
API catalog bảng và cột dành cho kiểm tra thủ công.
API verify nhận cấu trúc bảng/cột/điều kiện, không nhận SQL từ FE.
Query compiler chỉ sinh SELECT, dùng PreparedStatement cho toàn bộ value.
API NDJSON stream, forward-only ResultSet, fetchSize, timeout và cancel khi client ngắt.
Chuẩn hóa response row thành phone, email, firstName, lastName, attributes.
Service-to-service authentication, correlation ID, metrics và health check.

API bàn giao

GET  /internal/v1/catalog/tables
GET  /internal/v1/catalog/tables/{schemaName}/{tableName}/columns
POST /internal/v1/audiences/verify
POST /internal/v1/audiences/stream
GET  /actuator/health
GET  /actuator/prometheus

Các trường hợp phải xử lý

Trường hợp	Xử lý
Table hoặc column không tồn tại	Trả error code, không mở stream.
Kiểu value không khớp column	Trả field error tương ứng.
Operator không hỗ trợ	Từ chối request.
Query timeout	Cancel statement, đóng stream, ghi log thời gian và requestId.
Client ngắt kết nối	Dừng đọc ResultSet và giải phóng connection.
Hadoop trả hàng triệu dòng	Stream liên tục, bộ nhớ không tăng theo tổng số dòng.
JDBC pool hết connection	Fail fast với mã lỗi quá tải, không chờ vô hạn.

Kết quả bàn giao

JAR, file cấu hình mẫu và hướng dẫn triển khai trong OCS.
Contract request/response OpenAPI.
Xác nhận driver, JDBC URL, fetchSize và timeout với team OCS.
Chứng minh bằng test rằng service stream dữ liệu lớn với bộ nhớ ổn định.

Task 2 - LaoAds BE quản lý danh mục bảng và cột OCS

Mục tiêu

Tạo nguồn cấu hình duy nhất trong LaoAds để FE chỉ nhìn thấy các bảng, cột và operator được phép sử dụng.

Phạm vi

Tạo model, repository, index và migration cho ocs_dataset_config và ocs_dataset_field_config.
Giữ cách trả label đa ngôn ngữ và metadata field giống user_lao_id_field_config, nhưng dùng DTO/model riêng cho OCS.
API dành cho FE lấy dataset ACTIVE và danh sách column.
API admin nội bộ để search, tạo, cập nhật, activate/inactivate dataset; chưa bắt buộc có màn FE admin.
Validate trùng datasetCode, trùng table; field phải unique theo (datasetCode, fieldCode) và (datasetCode, columnName).
Validate phải có ít nhất một field ánh xạ PHONE hoặc EMAIL nếu dataset dùng làm audience gửi tin.
Cache danh mục ngắn hạn nếu cần; update phải xóa đúng cache key.
MapStruct cho model, request và response.

MongoDB thay đổi

New: ocs_dataset_config
New: ocs_dataset_field_config

API bàn giao

GET  /ocs-datasets
GET  /ocs-datasets/{datasetCode}/columns
POST /admin/ocs-datasets/search
POST /admin/ocs-datasets
PUT  /admin/ocs-datasets/{datasetCode}
POST /admin/ocs-datasets/{datasetCode}/activate
POST /admin/ocs-datasets/{datasetCode}/inactivate
POST /admin/ocs-datasets/{datasetCode}/fields
PUT  /admin/ocs-datasets/{datasetCode}/fields/{fieldCode}
POST /admin/ocs-datasets/{datasetCode}/fields/{fieldCode}/activate
POST /admin/ocs-datasets/{datasetCode}/fields/{fieldCode}/inactivate

Luồng cập nhật schema

Kỹ sư gọi hai API catalog của OCS để kiểm tra thủ công.
Kỹ sư cập nhật cấu hình bằng migration hoặc API admin LaoAds.
LaoAds validate và lưu Mongo.
Cache danh mục được xóa.
FE tải lại sẽ thấy dữ liệu mới.

Kết quả bàn giao

FE không phụ thuộc trực tiếp vào metadata Hadoop.
Thêm hoặc ẩn table/column không cần build lại JAR OCS.
Company user không thể chọn field ngoài danh mục được cấp.

Task 3 - LaoAds BE chuyển Segment QUERY từ LaoID sang OCS

Mục tiêu

Thay luồng QUERY hiện tại bằng định nghĩa OCS có cấu trúc, verify trước khi lưu và giữ nguyên kiến trúc Segment/SegmentVersion.

Phạm vi

Bổ sung queryDefinition vào SegmentModel, SegmentSourceSnapshotModel, DTO và response.
Tạo enum dùng chung cho data type, target field, operator, value source và relation.
Bổ sung Bean Validation và business validation theo ocs_dataset_config và ocs_dataset_field_config.
Resolve fieldCode sang column vật lý ở backend; không tin schema/table/column từ FE.
Gọi POST /internal/v1/audiences/verify khi tạo hoặc sửa Segment.
Lưu definitionChecksum và verifiedAt sau verify thành công.
Query Segment luôn được backend gán segmentType = DYNAMIC, refreshMode = PER_RUN.
Tạo Segment không query dữ liệu và không tạo SegmentVersion ngay.
Approval Campaign chỉ kiểm tra Segment active và cấu hình hợp lệ; không yêu cầu QUERY Segment đã có version READY.
Loại bỏ đường gọi LaoID CustomerQueryBuilder -> JdbcTemplate -> user_basic_info khỏi Segment QUERY.
Dùng MapStruct; không map tay request/response.

Code hiện tại bị ảnh hưởng

SegmentModel
SegmentSourceSnapshotModel
CreateSegmentRequest / Update Segment request
SegmentDto
SegmentService
CampaignService validation
CustomerQueryBuilder
UserFieldController và user_lao_id_field_config sau khi kiểm tra dependency

Các trường hợp phải xử lý

Trường hợp	Xử lý
Dataset `INACTIVE`	Không cho tạo/sửa Segment mới.
Field không thuộc dataset	Trả lỗi đúng field.
Field không selectable nhưng nằm trong selectedFields	Từ chối.
Field không filterable nhưng được dùng làm condition	Từ chối.
Thiếu required filter	Từ chối để tránh full scan.
Khoảng ngày vượt giới hạn	Từ chối và trả giới hạn cho FE.
Không có PHONE/EMAIL	Từ chối vì không tạo được recipient.
OCS verify timeout	Không lưu cấu hình mới; Segment cũ không bị thay đổi.
Sửa Segment sau khi đã có Run	Run cũ giữ snapshot; Run mới dùng định nghĩa mới.
Segment QUERY test cũ	Chỉnh sửa và verify lại hoặc xóa rồi tạo mới; không tự diễn giải điều kiện LaoID thành OCS.

Kết quả bàn giao

Segment QUERY mới không còn phụ thuộc database LaoID.
Không có raw SQL do FE tạo.
ErrorCode có message đa ngôn ngữ và field error đủ để FE hiển thị.
FILE và EVENT không bị thay đổi hành vi.

Task 4 - FE tạo và xem Segment nguồn OCS

Mục tiêu

Cho người dùng tạo Segment OCS bằng các điều khiển rõ ràng mà không cần biết SQL hoặc tên cột vật lý.

Phạm vi

Màn tạo/sửa Segment QUERY gọi danh sách dataset từ LaoAds.
Khi chọn dataset, tải column và operator tương ứng.
Chọn nhiều output field, nhưng bắt buộc có phone hoặc email theo metadata BE trả về.
Dựng input đúng dataType: text, number, date, date range hoặc multi-select.
Hỗ trợ FIXED và RUN_DATE cho field thời gian được backend cho phép.
Hiển thị field error theo errorCode của backend.
Màn detail hiển thị định nghĩa query và lịch sử SegmentVersion.
API member tiếp tục đọc theo version; không gọi API segment/user/{segmentId} cũ.

Các trường hợp phải xử lý

Trường hợp	Xử lý FE
Dataset bị inactivate khi form đang mở	Submit lỗi thì reload catalog và giữ các field còn hợp lệ.
Đổi dataset	Xóa selected field và condition của dataset cũ sau khi người dùng xác nhận.
Segment chưa có version	Hiển thị “Chưa có lần build dữ liệu”, không coi là lỗi.
Version BUILDING	Hiển thị tiến độ và tự refresh có giới hạn.
Version FAILED	Hiển thị mã lỗi và thời điểm lỗi; không hiển thị dữ liệu attempt dở.

Kết quả bàn giao

Người dùng hoàn thành luồng: chọn bảng -> chọn cột -> lọc -> verify/lưu.
Không có ô nhập SQL.
Màn detail phân biệt rõ định nghĩa Segment và dữ liệu của từng version.

Task 5 - Segment Import Worker build OCS SegmentVersion

Mục tiêu

Đọc NDJSON từ OCS, ghi ClickHouse theo batch và đảm bảo một build có thể retry an toàn trong môi trường nhiều node.

Phạm vi

Tạo consumer cho yêu cầu build OCS SegmentVersion.
Tạo segment_build_job, unique buildKey, lease và heartbeat.
Gọi OCS stream bằng RestClient hỗ trợ đọc line-by-line; không deserialize toàn bộ body.
Validate record START, DATA, END, ERROR.
Normalize phone/email, tạo recipientKey, validate member.
Insert ClickHouse theo batch mặc định 1.000 dòng.
Dùng importAttemptId cho từng attempt.
Cập nhật progress có throttle; log theo job/version/attempt.
Retry hữu hạn và chỉ đánh dấu version READY sau khi nhận END, flush hết batch và đối soát count.
Publish trạng thái hoàn thành/thất bại để CampaignRun tiếp tục hoặc kết thúc.

Dữ liệu thay đổi

New MongoDB: segment_build_job
Alter MongoDB: segment_version.successfulAttemptId
Alter ClickHouse: segment_version_member.import_attempt_id
Alter ClickHouse: segment_version_member.source_data_as_of

Các trường hợp phải xử lý

Trường hợp	Xử lý
Hai worker nhận cùng message	Chỉ worker claim được lease mới chạy.
Kafka gửi lại message	Unique buildKey và trạng thái job làm request idempotent.
Dòng thiếu cả phone và email	Ghi invalid count; không insert target member.
Trùng recipient trong một version	Dedupe theo `recipientKey` theo chiến lược đã chốt.
ClickHouse lỗi giữa batch	Không ack thành công; attempt bị fail và retry với attemptId mới.
OCS stream kết thúc không có END	Attempt thất bại.
Worker chết	Lease hết hạn để node khác claim và chạy attempt mới.
Attempt cũ có dữ liệu trong ClickHouse	Không được đọc vì khác `successfulAttemptId`; cleanup sau.

Kết quả bàn giao

Dữ liệu lớn không làm RAM tăng theo tổng số member.
Không có hai successful build cho cùng buildKey.
Có thể truy vết bằng requestId, buildJobId, segmentVersionId và importAttemptId.

Task 6 - CampaignRun orchestration cho Segment OCS

Mục tiêu

Đảm bảo mỗi lần Campaign chạy chờ đúng SegmentVersion, chốt đúng TargetSnapshot và không dispatch dữ liệu cũ.

Phạm vi

Khi đến lịch, tạo CampaignRun với unique scheduleOccurrenceKey.
Chuyển run sang PREPARING_TARGET.
Snapshot định nghĩa Segment và resolve RUN_DATE theo timezone Campaign/company.
Tạo buildKey; tạo mới hoặc dùng chung SegmentVersion đang build/đã READY phù hợp.
Chờ build hoàn thành bằng event hoặc polling job có giới hạn; không block scheduler thread.
Nếu version READY, tạo TargetSnapshot và campaign_target_member riêng cho run.
Chuyển run sang READY_TO_DISPATCH rồi mới gửi Kafka dispatch.
Nếu build thất bại, run chuyển FAILED với errorCode từ build job.
Kết thúc sớm Campaign không cho tạo run mới và hủy run chưa dispatch theo state machine hiện tại.

Các trường hợp phải xử lý

Trường hợp	Xử lý
222 Campaign cùng Segment và cùng sourceDataAsOf	Dùng chung một SegmentVersion, mỗi Run có TargetSnapshot riêng.
Cùng Segment nhưng chạy ngày khác	Tạo version khác do buildKey khác.
Segment được sửa trong lúc version đang build	Build hiện tại giữ snapshot cũ; run sau dùng định nghĩa mới.
Segment bị archive sau khi Run đã PREPARING_TARGET	Run đã snapshot có thể tiếp tục theo chính sách; không tạo Run mới.
OCS chậm	Run ở PREPARING_TARGET, hiển thị progress; không dispatch version cũ thay thế.
Scheduler chạy trên nhiều node	Unique scheduleOccurrenceKey và buildKey ngăn tạo trùng.

Kết quả bàn giao

Không còn trường hợp Campaign interval lấy mù latestReadyVersionId của lần trước.
Báo cáo và delivery luôn gắn được với campaignRunId, targetSnapshotId và segmentVersionId.
Một lỗi OCS không làm scheduler thread treo hoặc tạo vòng lặp vô hạn.

Task 7 - Cấu hình triển khai, bảo mật và theo dõi vận hành

Mục tiêu

Chuẩn hóa cấu hình để hệ thống chạy ổn định giữa hai vùng LaoAds và OCS, đồng thời đủ dữ liệu để phát hiện và truy vết sự cố.

Phạm vi

Xác nhận engine JDBC, driver, URL, authentication và timezone với OCS.
Cấu hình pool nhỏ có giới hạn, query timeout, socket timeout và max concurrent query.
Network allowlist, TLS/mTLS hoặc service credential giữa LaoAds và OCS.
Không log phone, email, token, credential hoặc condition value nhạy cảm.
Log bắt buộc: requestId, companyId, segmentId, versionId, buildJobId, attemptId, durationMs, rowCount, errorCode.
Metrics: active query, queued query, stream duration, rows streamed, bytes streamed, worker batch latency, retry count, failed build.
Alert cho tỷ lệ build fail, query timeout, queue backlog, lease timeout và ClickHouse insert fail.
Runbook kiểm tra catalog, verify, stream, retry và rollback JAR.

Kết quả bàn giao

Có dashboard và cảnh báo tối thiểu trước khi chạy production.
Có hướng dẫn kiểm tra thủ công schema OCS và cập nhật Mongo catalog.
Có giới hạn tài nguyên rõ ràng để một query lớn không chiếm hết OCS Audience Service.

12. Thứ tự triển khai

[Diagram]

Thứ tự đề xuất:

Chốt engine JDBC và contract OCS trước khi code tích hợp.
Task 1 và Task 2 có thể chạy song song.
Task 3 bắt đầu khi đã chốt contract verify cùng cấu trúc ocs_dataset_config và ocs_dataset_field_config.
Task 4 có thể ghép mock API sau khi Task 2 chốt response.
Task 5 tích hợp stream thật sau khi Task 1 có môi trường test.
Task 6 chỉ ghép hoàn chỉnh khi Task 5 đảm bảo version READY/FAILED chính xác.
Task 7 đi cùng toàn bộ quá trình, không để đến cuối mới bổ sung log và bảo mật.

13. Mã lỗi cần bổ sung

Tên code có thể điều chỉnh theo registry hiện tại, nhưng phải tách rõ lỗi cấu hình, lỗi OCS và lỗi build:

OCS_DATASET0001  Dataset không tồn tại
OCS_DATASET0002  Dataset không hoạt động
OCS_DATASET0003  Field không thuộc dataset
OCS_DATASET0004  Field không được phép select
OCS_DATASET0005  Field không được phép filter
OCS_DATASET0006  Operator không hợp lệ
OCS_DATASET0007  Thiếu điều kiện bắt buộc
OCS_DATASET0008  Khoảng dữ liệu vượt giới hạn

OCS_VERIFY0001   OCS verify thất bại
OCS_VERIFY0002   Table hoặc column thực tế không tồn tại
OCS_VERIFY0003   Kiểu dữ liệu không khớp
OCS_VERIFY0004   OCS verify timeout

OCS_STREAM0001   Không thể mở stream
OCS_STREAM0002   Stream kết thúc không đầy đủ
OCS_STREAM0003   Dữ liệu stream không đúng contract
OCS_STREAM0004   Hadoop query bị gián đoạn

SEGMENT_BUILD0001  Build job không tồn tại
SEGMENT_BUILD0002  Không thể claim build job
SEGMENT_BUILD0003  ClickHouse insert thất bại
SEGMENT_BUILD0004  Build vượt số lần retry
SEGMENT_BUILD0005  Count đối soát không khớp

Message trả về lấy theo StatusCodeEnum và file đa ngôn ngữ. Không hard-code message nghiệp vụ trong service nếu error code đã có message tương ứng.

14. Kết luận thiết kế

Phương án cuối cùng không dùng Query Template và không dùng GraphQL. Độ linh hoạt được cung cấp bằng danh mục bảng/cột có kiểm soát trong MongoDB và cấu trúc điều kiện cố định.

Ranh giới quan trọng:

FE chỉ biết datasetCode, fieldCode, operator và value.
LaoAds Mongo quyết định người dùng được nhìn thấy và sử dụng bảng/cột nào.
Hai API catalog OCS chỉ để kỹ sư kiểm tra thủ công, không nằm trong request path của FE.
OCS verify và OCS stream mới là hai API LaoAds/Worker sử dụng trong luồng thật.
OCS Service quyết định câu query kỹ thuật cuối cùng và chỉ dùng bind parameter.
Mongo lưu metadata/trạng thái; ClickHouse lưu member số lượng lớn; Hadoop giữ dữ liệu nguồn.
Campaign chỉ dispatch sau khi SegmentVersion READY và TargetSnapshot đã được chốt.

Thiết kế này cho phép OCS thay đổi bảng/cột mà không phải sửa giao diện hoặc build lại JAR cho mỗi thay đổi nhỏ, đồng thời vẫn giữ quyền kiểm soát dữ liệu và hiệu suất trong LaoAds.

OCS Audience Service và luồng Segment nguồn OCS

1. Mục tiêu

2. Phạm vi thay đổi

2.1. Phần giữ nguyên

2.2. Phần loại bỏ khỏi luồng Segment QUERY

2.3. Các thành phần mới

3. Kiến trúc tổng thể

3.1. Trách nhiệm của từng hệ thống

4. Luồng giao diện

4.1. API FE sử dụng để tạo Segment

4.2. Các trường FE cần gửi

4.3. Màn chi tiết Segment

5. Danh mục bảng và cột OCS trong MongoDB

5.1. Collection ocs_dataset_config

5.2. Collection ocs_dataset_field_config

5.3. So sánh với user_lao_id_field_config

5.4. Ý nghĩa các trường

5.5. Index MongoDB

5.6. Cách cập nhật khi OCS thay đổi schema

6. Dữ liệu lưu trong segment

6.1. Bổ sung queryDefinition

6.2. SegmentSourceSnapshotModel

6.3. segment_version

7. Hợp đồng OCS Audience Service

7.1. API catalog dùng để kiểm tra thủ công

7.2. API verify được LaoAds gọi

7.3. API stream được Worker gọi

7.4. Định dạng NDJSON

7.5. Tại sao stream không làm tràn RAM

8. Luồng tạo và sửa Segment

9. Luồng CampaignRun và build SegmentVersion

9.1. buildKey và trường hợp nhiều Campaign dùng chung Segment

9.2. Collection segment_build_job

9.3. Bổ sung ClickHouse segment_version_member

10. Xử lý lỗi và retry

11. Phân chia task

Task 1 - Xây dựng OCS Audience Service

Task 2 - LaoAds BE quản lý danh mục bảng và cột OCS

Task 3 - LaoAds BE chuyển Segment QUERY từ LaoID sang OCS

Task 4 - FE tạo và xem Segment nguồn OCS

Task 5 - Segment Import Worker build OCS SegmentVersion

Task 6 - CampaignRun orchestration cho Segment OCS

Task 7 - Cấu hình triển khai, bảo mật và theo dõi vận hành

12. Thứ tự triển khai

13. Mã lỗi cần bổ sung

14. Kết luận thiết kế

5.1. Collection `ocs_dataset_config`

5.2. Collection `ocs_dataset_field_config`

5.3. So sánh với `user_lao_id_field_config`

6. Dữ liệu lưu trong `segment`

6.1. Bổ sung `queryDefinition`

6.2. `SegmentSourceSnapshotModel`

6.3. `segment_version`

9.1. `buildKey` và trường hợp nhiều Campaign dùng chung Segment

9.2. Collection `segment_build_job`

9.3. Bổ sung ClickHouse `segment_version_member`