syntax = "proto2";

package whpb;

import "gogoproto/gogo.proto";

option (gogoproto.marshaler_all) = true;
option (gogoproto.sizer_all) = true;
option (gogoproto.unmarshaler_all) = true;

message Label {
    repeated string vals    = 1;
}

// mysql_binlog and mysql_dump is for Portal.
// mysql_sync is for stargate.
enum SourceType {
    mysql_binlog    =  0;
    mysql_dump      =  1;
}

enum WorkerType {
    Portal    = 0;
    Stargate  = 1;
}

message Worker {
    optional uint64 id = 1 [(gogoproto.nullable) = false];
    optional WorkerType type = 2 [(gogoproto.nullable) = false];

    // labels of worker like group, support protocol etc.
    map<string, Label> labels = 3;

    // some annotations
    map<string, string> annotations     = 4;
    optional int64 create_time          = 5 [(gogoproto.nullable) = false];
    optional int64 last_time            = 6 [(gogoproto.nullable) = false];
}

enum OPType {
    DDL     = 0;
    DML     = 1;
    END     = 2;
}

enum EventType {
    Insert = 0;
    Update = 1;
    Delete = 2;
}

enum JobState {
    Normal     = 0;
    Abnormal   = 1;
    Paused     = 2;
}

enum JobStage {
    New           = 0;
    Initializing  = 1;
    Running       = 2;
    Resolved      = 3;
    Deleting      = 4;
    Deleted       = 5;
}

// job's progress position information.
message Position {
    optional string mysql_binlog_file  = 1 [(gogoproto.nullable) = false];
    optional uint64 mysql_binlog_pos   = 2 [(gogoproto.nullable) = false];
    optional string mysql_binlog_gtid  = 3 [(gogoproto.nullable) = false];
    optional int64 mongo_ts            = 4 [(gogoproto.nullable) = false];
    optional int64 kafka_offset        = 5 [(gogoproto.nullable) = false];
}

message JobStatus {
    optional uint64 id                  = 1 [(gogoproto.nullable) = false];
    optional JobStage stage             = 2 [(gogoproto.nullable) = false];
    optional JobState state             = 3 [(gogoproto.nullable) = false];
    optional string type                = 4 [(gogoproto.nullable) = false];
    optional int32 replicate            = 5 [(gogoproto.nullable) = false];

    // job's total quantity
    optional uint64 total               = 6 [(gogoproto.nullable) = false];
    // job's progress expressed by percent
    optional uint64 percent             = 7 [(gogoproto.nullable) = false];
    // job's current progress.
    optional Position pos               = 8;
    // job's qps
    optional uint64 qps                 = 9 [(gogoproto.nullable) = false];
    // The number of messages that not consumed
    optional int64 delay_message_count  = 10;
    // error message
    optional string message             = 11 [(gogoproto.nullable) = false];
}

message Job {
    optional uint64 id                  = 1 [(gogoproto.nullable) = false];
    optional JobStatus  status          = 2;
    optional WorkerType type            = 3 [(gogoproto.nullable) = false];
    optional SourceType source          = 4 [(gogoproto.nullable) = false];
    // detail information of job, contains source/target db and mq topic informations.
    optional bytes info                 = 5;
     // labels of job, used to schedule job and worker.
    map<string, Label> labels           = 6;
    // some annotations, cluster etcd root path stored in `clusterRootPath`.
    map<string, string> annotations     = 7;
    // dependencies of job, scheduled util all those jobs are finished.
    repeated uint64 dependencies        = 8;
}

message MysqlDumpJob {
    // datasource information
    optional DBAccount source       = 1 [(gogoproto.nullable) = false];
    // optional. databases and tables to dump
    repeated Schema schemas         = 2;
    // for later scale use
    optional int32 partition_num    = 3 [(gogoproto.nullable) = false];
}

message MysqlBinlogJob {
    // datasource information
    optional DBAccount source     = 1  [(gogoproto.nullable) = false];
    // optional. databases and tables to fetch binlog.
    repeated Schema schemas       = 2;
    // for later scale use
    optional int32 partition_num  = 3 [(gogoproto.nullable) = false];
    optional int32 server_id      = 4 [(gogoproto.nullable) = false];
}

message MysqlSyncJob {
    // datasource information.
    optional SchemaInfo source     = 1  [(gogoproto.nullable) = false];
    // target database information.
    optional DBAccount target      = 2 [(gogoproto.nullable) = false];
    // optional. sharding rules.
    repeated RouteRule rules       = 3;
}

message DBAccount {
    optional string host            = 1 [(gogoproto.nullable) = false];
    optional int32 port             = 2 [(gogoproto.nullable) = false];
    optional string username        = 3 [(gogoproto.nullable) = false];
    optional string password        = 4 [(gogoproto.nullable) = false];
}

// for pattern, we use simple wildcard:
// 1. asterisk character (*, also called "star") matches zero or more characters,
//    for example, doc* matches doc and document but not dodo;
//    asterisk character must be in the end of wildcard word,
//    and there is only one asterisk in one wildcard word
// 2. the question mark ? matches exactly one character
message RouteRule {
    optional string schema_pattern      = 1 [(gogoproto.nullable) = false];
    optional string table_pattern       = 2;
    optional string target_schema       = 3 [(gogoproto.nullable) = false];
    optional string target_table        = 4;
}

message SchemaInfo {
    optional string host       = 1 [(gogoproto.nullable) = false];
    optional int32  port       = 2 [(gogoproto.nullable) = false];
    repeated Schema schemas    = 3;
}

message Schema {
    optional string name        = 1 [(gogoproto.nullable) = false];
    repeated string tables      = 2;
    optional bool noLocks       = 3 [(gogoproto.nullable) = false];
    optional bool willSyncMore  = 4 [(gogoproto.nullable) = false];
}

message SchemaDetail {
    // detail in JSON format.
    optional bytes detail  = 1;
    optional string schema = 2 [(gogoproto.nullable) = false];
    optional string table = 3 [(gogoproto.nullable) = false];
}

message Topic {
    optional string name         = 1 [(gogoproto.nullable) = false];
    optional string host         = 2 [(gogoproto.nullable) = false];
    optional int32  port         = 3 [(gogoproto.nullable) = false];
    optional string schema       = 4 [(gogoproto.nullable) = false];
    optional string table        = 5 [(gogoproto.nullable) = false];
    optional int32 partition_num = 6 [(gogoproto.nullable) = false];
}

message Partition {
    optional string topic_name      = 1 [(gogoproto.nullable) = false];
    optional int32  partition_id    = 2 [(gogoproto.nullable) = false];
    optional Position pos           = 3 ;
}

message Partitions {
    repeated Partition partitions = 1;
}

message TopicPartitionBitmap {
    optional Topic topic                    = 1;
    repeated uint64 partition_bitmap        = 2;
    repeated uint64 partition_midway_bitmap = 3;
}

// Fence is a primitive to sync all consumers to consume old messages firstly
// use bitmap to mark which partition is already finished consuming [0,...,max patition ID]
message Fence {
    optional uint64 id                      = 1 [(gogoproto.nullable) = false];
    repeated TopicPartitionBitmap bitmaps   = 2;
}

message DMLData {
    optional string schema            = 1 [(gogoproto.nullable) = false];
    optional string table             = 2 [(gogoproto.nullable) = false];
    optional EventType tp             = 3 [(gogoproto.nullable) = false];
    repeated RowData  rows            = 4;
    repeated Column columns           = 5;
}

message RowData {
    repeated string row = 1;
}

message Column {
    optional int32 idx      = 1 [(gogoproto.nullable) = false];
    optional string name    = 2 [(gogoproto.nullable) = false];
    optional bool isKey     = 3 [(gogoproto.nullable) = false];
    optional int32 type     = 4 [(gogoproto.nullable) = false];
}

message Drip {
    optional string schema          = 1 [(gogoproto.nullable) = false];
    optional string table           = 2 [(gogoproto.nullable) = false];
    optional string key             = 3 [(gogoproto.nullable) = false];
    optional SourceType source      = 4 [(gogoproto.nullable) = false];
    optional OPType op              = 5 [(gogoproto.nullable) = false];
    // dml data in this field of DML Drip event.
    optional bytes  data            = 6;
    // fence id if op == DDL, fence.
    // ddl is a special fence, while we meet ddl, we should wait util all consumers reach all fence.
    // then we can agree to a person to do this ddl.
    optional uint64 fence_id        = 7 [(gogoproto.nullable) = false];
    // position of Drip event in log event
    optional Position pos           = 8;
    // create time of Drip event.
    optional int64 create_time      = 9 [(gogoproto.nullable) = false];
}