syntax = "proto3";

package materialize;
option go_package = "github.com/estuary/flow/go/protocols/materialize";

import "consumer/protocol/protocol.proto";
import "go/protocols/flow/flow.proto";
import "gogoproto/gogo.proto";
import "google/protobuf/timestamp.proto";

option (gogoproto.marshaler_all) = true;
option (gogoproto.protosizer_all) = true;
option (gogoproto.unmarshaler_all) = true;
option (gogoproto.goproto_getters_all) = false;

service Connector {
  rpc Materialize(stream Request) returns (stream Response);
}

// Request of a materialize RPC.
//
// Exactly one sub-message field of Request is set, optionally alongside
// `internal`. The Rust bindings enforce that XOR as an injected `oneof`, driven
// by an explicit field list which MUST be extended when a sub-message field is
// added below: see `ONEOFS` in crates/proto-flow/build.rs, which also explains
// why this is a Rust-only concern.
message Request {
  // Spec requests the specification definition of this connector.
  // Notably this includes its configuration JSON schemas.
  message Spec {
    // Connector type addressed by this request.
    flow.MaterializationSpec.ConnectorType connector_type = 1;
    // Connector configuration, as an encoded JSON object.
    // This may be a partial specification (for example, a Docker image),
    // providing only enough information to fetch the remainder of the
    // specification schema.
    bytes config_json = 2 [
      (gogoproto.casttype) = "encoding/json.RawMessage",
      json_name = "config"
    ];
  }
  Spec spec = 1;

  // Validate a materialization configuration and proposed bindings.
  // Validate is run out-of-band with ongoing capture invocations.
  // It's purpose is to confirm that the proposed configuration
  // is likely to succeed if applied and run, or to report any
  // potential issues for the user to address.
  message Validate {
    // Name of the materialization being validated.
    string name = 1
        [ (gogoproto.casttype) =
              "github.com/estuary/flow/go/protocols/flow.Materialization" ];
    // Connector type addressed by this request.
    flow.MaterializationSpec.ConnectorType connector_type = 2;
    // Connector configuration, as an encoded JSON object.
    bytes config_json = 3 [
      (gogoproto.casttype) = "encoding/json.RawMessage",
      json_name = "config"
    ];
    // Bindings of endpoint resources and collections from which they would be
    // materialized. Bindings are ordered and unique on their resource path.
    // Multiple bindings may materialize from a single collection.
    message Binding {
      // JSON-encoded object which specifies the endpoint resource to be materialized.
      bytes resource_config_json = 1 [
        (gogoproto.casttype) = "encoding/json.RawMessage",
        json_name = "resourceConfig"
      ];
      // Collection to be materialized.
      // Unset if the request uses `linked_collections`, in which case
      // `collection_index` identifies the bound collection instead.
      flow.CollectionSpec collection = 2 [ (gogoproto.nullable) = false ];
      // Projection configuration, keyed by the projection field name,
      // with JSON-encoded and driver-defined configuration objects.
      map<string, bytes> field_config_json_map = 3 [
        (gogoproto.castvalue) = "encoding/json.RawMessage",
        json_name = "fieldConfig"
      ];
      // Backfill counter for this binding.
      uint32 backfill = 4;
      // Ordered group-by key fields of the materialization.
      // Identical to `keys` of the post-validation FieldSelection.
      repeated string group_by = 5;
      // Index of this binding's collection within the request's
      // `linked_collections`. Used instead of `collection` when the request is
      // in indirect form, and unset otherwise.
      uint32 collection_index = 6;
    }
    repeated Binding bindings = 4;
    // Last MaterializationSpec which was validated and published.
    // Note that this MaterializationSpec may not have been applied.
    flow.MaterializationSpec last_materialization = 5;
    // Version of the last validated MaterializationSpec.
    string last_version = 6;
    // Table of collections referenced by `collection_index` of this request's
    // bindings, in place of an inlined `collection` (the "indirect" form).
    //
    // The form is a property of the request as a whole: if
    // `linked_collections` is non-empty then every binding leaves `collection`
    // unset and resolves through `collection_index`. If it's empty then every
    // binding inlines its `collection` and `collection_index` is unset.
    // Mixed forms are invalid. `last_materialization` is self-contained and
    // carries whichever form it was built with, independent of this request.
    //
    // Entries have `derivation` cleared, as inlined collections always do.
    // Builders emit entries which are unique by value and ordered on name,
    // but that's a convention only and not an invariant of the message.
    // Duplicate names are legal and do arise here: bindings of one collection
    // having differing `group_by` rewrite the binding's `key` and
    // `projections[*].is_primary_key` to match, so same name != same spec.
    repeated flow.CollectionSpec linked_collections = 7
        [ (gogoproto.nullable) = false ];
    // Secrets of this materialization, mapping a catalog secret name to its JSON
    // pointer within `config_json`. The runtime resolves and merges each secret
    // into that location, as an RFC 7396 merge patch, before handing the
    // configuration to the connector.
    map<string, string> secrets = 8;
  }
  Validate validate = 2;

  // Apply an updated materialization specification to its endpoint,
  // in preparation for an Open of a materialization session.
  // Apply is run by the leader shard of a materialization task
  // (having key_begin: 0) while the materialization is quiescent.
  // Apply may be called multiple times for a given `version` and
  // `last_version`, even if a prior call succeeded from the connector's
  // perspective, so implementations must be idempotent. However, the next
  // session will not Open until it's preceding Apply has durably completed.
  message Apply {
    // Materialization to be applied.
    flow.MaterializationSpec materialization = 1;
    // Version of the MaterializationSpec being applied.
    string version = 2;
    // Last CaptureSpec which was successfully applied.
    flow.MaterializationSpec last_materialization = 4;
    // Version of the last applied MaterializationSpec.
    string last_version = 5;
    // Last-persisted connector checkpoint state from a previous session.
    // The Apply RPC may use this state to perform a post-commit apply
    // of files staged under the `last_materialization` specification.
    bytes state_json = 6 [
      (gogoproto.casttype) = "encoding/json.RawMessage",
      json_name = "state"
    ];
  }
  Apply apply = 3;

  // Open a materialization stream.
  //
  // If the Flow recovery log is authoritative:
  // The driver is given its last committed checkpoint state in this request.
  // It MAY return a runtime checkpoint in its opened response -- perhaps an older
  // Flow checkpoint which was previously embedded within its driver checkpoint.
  //
  // If the remote store is authoritative:
  // The driver MUST fence off other streams of this materialization that
  // overlap the provided [key_begin, key_end) range, such that those streams
  // cannot issue further commits. The driver MUST return its stored runtime
  // checkpoint for this materialization and range [key_begin, key_end]
  // in its Opened response.
  //
  // After Open, the runtime will send only Load, Flush, Store,
  // StartCommit, and Acknowledge.
  message Open {
    // Materialization to be transacted.
    flow.MaterializationSpec materialization = 1;
    // Version of the opened MaterializationSpec.
    // The driver may want to require that this match the version last
    // provided to a successful Apply RPC. It's possible that it won't,
    // due to expected propagation races in Flow's distributed runtime.
    string version = 2;
    // Range of documents to be processed by this session.
    flow.RangeSpec range = 3;
    // Last-persisted connector checkpoint state from a previous session.
    bytes state_json = 4 [
      (gogoproto.casttype) = "encoding/json.RawMessage",
      json_name = "state"
    ];
    // Endpoint configuration of this materialization exactly as published,
    // before whole-document unsealing or secrets-stanza injection. Connectors
    // may reuse it as the non-secret baseline of a `configUpdate`.
    bytes sealed_config_json = 5 [
      (gogoproto.casttype) = "encoding/json.RawMessage",
      json_name = "sealedConfig"
    ];
  }
  Open open = 4;

  // Load a document identified by its key. The given key may have never before been stored,
  // but a given key will be sent in a transaction Load just one time.
  //
  // The runtime populates exactly one of `key_json` or `key_packed` per the
  // negotiated codec.
  message Load {
    // Index of the Open binding for which this document is to be loaded.
    uint32 binding = 1;
    // Key tuple, as a JSON array of key components.
    // Ordering matches `keys` of the materialization's field selection.
    // Set only for JSON connectors; empty otherwise.
    bytes key_json = 2 [
      (gogoproto.casttype) = "encoding/json.RawMessage",
      json_name = "key"
    ];
    // Packed FoundationDB tuple of the document key to load.
    // Set only for protobuf connectors; empty otherwise.
    bytes key_packed = 3;
  }
  Load load = 5;

  // Flush loads. No further Loads will be sent in this transaction,
  // and the runtime will await the connectors's remaining Loaded
  // responses followed by one Flushed response.
  message Flush {
    // Aggregated state patches from all shards' prior-transaction Acknowledged
    // responses, as a tab-delimited JSON array. Includes this shard's own
    // patches (the runtime feeds the shard's contribution back to it for
    // symmetry with the scaled-out case). Connectors participating in
    // cooperative multi-shard strategies use this to observe peers' state.
    bytes state_patches_json = 1 [
      (gogoproto.casttype) = "encoding/json.RawMessage",
      json_name = "statePatches"
    ];

    // Backfill-begin signals observed during this transaction. A connector
    // acts on those relevant to its key range.
    message BackfillBegin {
      uint32 binding = 1;
      // Truncation boundary of the binding's backfill: documents published at or
      // after this time are current, while earlier ones were superseded by the
      // backfill. It equals the begin's own publication time (flow_published_at),
      // and is carried on both begin and complete so connectors need not track it
      // across transactions.
      google.protobuf.Timestamp timestamp = 2;
    }
    repeated BackfillBegin backfill_begins = 2;

    // Backfill-complete signals observed during this transaction. See
    // BackfillBegin.
    message BackfillComplete {
      uint32 binding = 1;
      // Truncation boundary of the completed backfill (see BackfillBegin.timestamp):
      // the connector may delete destination rows whose flow_published_at predates
      // this time, as they were superseded by the backfill.
      google.protobuf.Timestamp timestamp = 2;
    }
    repeated BackfillComplete backfill_completes = 3;
  }
  Flush flush = 6;

  // Store documents updated by the current transaction.
  //
  // The runtime populates exactly one of the JSON encodings (`key_json`,
  // `values_json`) or the packed encodings (`key_packed`, `values_packed`),
  // per the negotiated codec. `doc_json` is independent of the
  // codec and is set whenever the document is stored.
  message Store {
    // Index of the Open binding for which this document is to be stored.
    uint32 binding = 1;
    // Key tuple, as a JSON array of key components.
    // Ordering matches `keys` of the materialization's field selection.
    // Set only for JSON connectors; empty otherwise.
    bytes key_json = 2 [
      (gogoproto.casttype) = "encoding/json.RawMessage",
      json_name = "key"
    ];
    // Packed FoundationDB tuple of the document key to store.
    // Set only for protobuf connectors; empty otherwise.
    bytes key_packed = 3;
    // Values tuple, as a JSON array of value components.
    // Ordering matches `values` of the materialization's field selection.
    // Set only for JSON connectors; empty otherwise.
    bytes values_json = 4 [
      (gogoproto.casttype) = "encoding/json.RawMessage",
      json_name = "values"
    ];
    // Packed FoundationDB tuple of the document values to store.
    // Set only for protobuf connectors; empty otherwise.
    bytes values_packed = 5;
    // JSON document to store.
    bytes doc_json = 6 [
        (gogoproto.casttype) = "encoding/json.RawMessage",
        json_name = "doc"
    ];
    // Exists is true if this document has previously been loaded or stored.
    bool exists = 7;
    // Delete is true if this document is being deleted, which also implies `exists`.
    bool delete = 8;
  }
  Store store = 7;

  // Mark the end of the Store phase, and if the remote store is authoritative,
  // instruct it to start committing its transaction.
  message StartCommit {
    // Flow runtime checkpoint to commit with this transaction.
    consumer.Checkpoint runtime_checkpoint = 1;
    // Aggregated state patches from all shards' current-transaction Flushed
    // responses, as a tab-delimited JSON array.
    bytes state_patches_json = 2 [
      (gogoproto.casttype) = "encoding/json.RawMessage",
      json_name = "statePatches"
    ];
  }
  StartCommit start_commit = 8;

  // Acknowledge to the connector that the previous transaction
  // has committed to the Flow runtime's recovery log.
  message Acknowledge {
    // Aggregated connector state patches from all shards' just-committed
    // StartedCommit responses, as a tab-delimited JSON array.
    bytes state_patches_json = 1 [
      (gogoproto.casttype) = "encoding/json.RawMessage",
      json_name = "statePatches"
    ];
  }
  Acknowledge acknowledge = 9;

  // Reserved for internal use.
  bytes internal = 100 [ json_name = "$internal" ];
}

// Response of a materialize RPC.
//
// Exactly one sub-message field of Response is set, optionally alongside
// `internal`. The Rust bindings enforce that XOR as an injected `oneof`, driven
// by an explicit field list which MUST be extended when a sub-message field is
// added below: see `ONEOFS` in crates/proto-flow/build.rs, which also explains
// why this is a Rust-only concern.
message Response {
  // Spec responds to Request.Spec.
  message Spec {
    // Protocol version must be 3032023.
    uint32 protocol = 1;
    // JSON schema of the connector's configuration.
    bytes config_schema_json = 2 [
      (gogoproto.casttype) = "encoding/json.RawMessage",
      json_name = "configSchema"
    ];
    // JSON schema of the connector's resource configuration.
    // Schemas must use specific annotations to communicate the expected
    // locations of injected platform variables:
    // - `x-collection-name: true`, a string location for a resource name (required).
    // - `x-schema-name: true`, a string location for a resource schema (optional).
    // - `x-delta-updates: true`, a boolean location for enabling delta-updates mode (optional).
    bytes resource_config_schema_json = 3 [
      (gogoproto.casttype) = "encoding/json.RawMessage",
      json_name = "resourceConfigSchema"
    ];
    // URL for connector's documentation.
    string documentation_url = 4;
    // Optional OAuth2 configuration.
    flow.OAuth2 oauth2 = 5;
    // Materializations don't use resource_path_pointers.
    // Instead, annotations are required of its resource_config_schema_json.
    reserved 6;
  }
  Spec spec = 1;

  // Validated responds to Request.Validate.
  message Validated {
    // ProjectionConstraint pairs a projection field name with a single Constraint.
    // A repeated list of ProjectionConstraint allows multiple constraints to be
    // expressed for the same field simultaneously. For example, INCOMPATIBLE and
    // LOCATION_REQUIRED on the same field signals that the field is required but
    // the existing destination column has an incompatible type, and a backfill is
    // needed.
    message ProjectionConstraint {
      string field = 1;
      Constraint constraint = 2;
    }

    // Constraint constrains the use of a flow.Projection within a materialization.
    message Constraint {
      // Type encodes a constraint type for this flow.Projection.
      enum Type {
        INVALID = 0;
        // A field with this specific name MUST be present.
        //
        // Connectors should return FIELD_REQUIRED for field names that have
        // special meaning to the function of the connector. For example, a
        // Slack connector might require a "text" field which is the body of
        // Slack messages to author.
        //
        // Connectors may return FIELD_REQUIRED constraints for fields which
        // were not present in the presented collection projections, in which
        // case the control plane will generate a suitable user-facing error.
        FIELD_REQUIRED = 1;
        // A field projection with this location pointer must be present.
        //
        // Connectors may return LOCATION_REQUIRED for the root JSON document
        // pointer within a standard materialization that stores the document
        // as a single column. Or alternatively, a connector may require
        // all top-level document properties and store each as separate
        // columns, which are merged together again in Loaded responses.
        LOCATION_REQUIRED = 2;
        // DEPRECATED. LOCATION_RECOMMENDED is equivalent to FIELD_OPTIONAL.
        LOCATION_RECOMMENDED = 3;
        // This field projection may be included in the materialization.
        //
        // Connectors must return FIELD_OPTIONAL for all fields which they
        // are capable of supporting.
        FIELD_OPTIONAL = 4;
        // This field projection cannot participate in the materialization.
        // This condition is permanent and cannot be resolved with a backfill.
        //
        // Connectors must return FIELD_FORBIDDEN for fields having
        // unsupported types, such as JSON `null`.
        FIELD_FORBIDDEN = 5;
        // This field projection cannot participate in the materialization.
        // However, a backfill of the binding would clear this condition
        // and the field could again participate.
        //
        // Connectors must return INCOMPATIBLE for fields which would be
        // FIELD_OPTIONAL if (and only if) the binding were to be backfilled.
        // A common use is for fields that map to an existing SQL column having
        // an incompatible type which cannot be migrated automatically.
        INCOMPATIBLE = 6;
        // DEPRECATED: UNSATISFIABLE is an alias for INCOMPATIBLE.
        // This variant is kept for backward compatibility with JSON-encoded
        // protobuf messages. New code should use INCOMPATIBLE instead.
        UNSATISFIABLE = 7;
      }
      Type type = 2;
      // User-facing reason for the given constraint. The control-plane will
      // supply additional context as to the collection, binding and field.
      //
      // Connectors are strongly encouraged to supply a rationale
      // for constraints other than FIELD_OPTIONAL.
      string reason = 3;
      // Some endpoints are unable to represent various field names due to
      // technical limitations on length, character set, or case sensitivity.
      // For these cases, the connector should provide a "folded" field name to
      // be used instead. Examples of folds include:
      // - Lower-casing
      // - Replacing `/` with `_`
      // - Mapping Unicode to ASCII via https://en.wikipedia.org/wiki/Punycode
      //
      // Folds may be lossy and result in duplicated folded field values.
      // That's okay. The control plane will ensure at most one field is
      // selected among a set of fields having identical folds.
      string folded_field = 4;
    }
    // Validation responses for each binding of the request, and matching the
    // request ordering. Each Binding must have a unique resource_path.
    message Binding {
      // When true, fields are case-insensitive within the materialized resource.
      // Fields that differ only in Unicode capitalization will conflict and the
      // control plane must ensure they are not co-selected.
      //
      // Many database systems are case-aware in their information schema, so we
      // want to preserve natural casing of fields, but they will still prohibit
      // (for example) "myField" and "MyField" from co-existing within a table.
      bool case_insensitive_fields = 5;
      // Field tag 1 was `constraints`, a `map<string, Constraint>` keyed by
      // field name. It was replaced by `projection_constraints`, which can
      // express multiple constraints for the same field.
      reserved 1;
      reserved "constraints";
      // Components of the resource path which fully qualify the resource
      // identified by this binding.
      // - For an RDBMS, this might be []{dbname, schema, table}.
      // - For Kafka, this might be []{topic}.
      // - For Redis, this might be []{key_prefix}.
      repeated string resource_path = 2;
      // Materialize combined delta updates of documents rather than full
      // reductions.
      //
      // When set, the Flow runtime will not attempt to load documents via
      // Request.Load, and also disables re-use of cached documents
      // stored in prior transactions. Each stored document is exclusively
      // combined from updates processed by the runtime within the current
      // transaction only.
      //
      // This is appropriate for drivers over streams, WebHooks, and append-only
      // files.
      //
      // For example, given a collection which reduces a sum count for each key,
      // its materialization will produce a stream of delta updates to the count,
      // such that a reader of the stream will arrive at the correct total count.
      bool delta_updates = 3;
      // Serialization policy to use for this binding.
      flow.SerPolicy ser_policy = 4;
      // Constraints on each projection, as a list that allows multiple constraints
      // per projection field. Projections of the CollectionSpec which are missing
      // from this list are implicitly forbidden.
      //
      // The list form allows expressing compound requirements that a single
      // constraint type cannot capture. For example, emitting both INCOMPATIBLE and
      // LOCATION_REQUIRED for the same field signals that the field is required
      // but the existing destination column is incompatible; a backfill is required.
      repeated ProjectionConstraint projection_constraints = 6;

      // Next tag: 7
    }
    repeated Binding bindings = 1;
  }
  Validated validated = 2;

  // Applied responds to Request.Apply.
  message Applied {
    // Human-readable description of the action that the connector took.
    // If empty, this Apply is to be considered a "no-op".
    string action_description = 1;
    // Optional *transactional* update to ConnectorState.
    // This update commits atomically with the Flow recovery log checkpoint
    // which marks the current specification as having been applied.
    flow.ConnectorState state = 2;
  }
  Applied applied = 3;

  // Opened responds to Request.Open.
  // After Opened, the connector sends only Loaded, Flushed,
  // StartedCommit, and Acknowledged as per the materialization
  // protocol.
  message Opened {
    // Flow runtime checkpoint to begin processing from.
    // If empty, the most recent checkpoint of the Flow recovery log is used.
    //
    // Or, a driver may send the value []byte{0xf8, 0xff, 0xff, 0xff, 0xf, 0x1}
    // to explicitly begin processing from a zero-valued checkpoint, effectively
    // rebuilding the materialization from scratch. This sentinel is a trivial
    // encoding of the max-value 2^29-1 protobuf tag with boolean true.
    consumer.Checkpoint runtime_checkpoint = 1;
    // Disable the runtime's load optimization which suppresses Load requests
    // for keys that are known to not exist in the endpoint. When set to true,
    // all Load requests will be sent to the connector regardless of the
    // maximum key tracking.
    bool disable_load_optimization = 2;
  }
  Opened opened = 4;

  // Loaded responds to Request.Load.
  // It returns documents of requested keys which have previously been stored.
  // Keys not found in the store MUST be omitted. Documents may be in any order,
  // both within and across Loaded response messages, but a document of a given
  // key MUST be sent at most one time in a Transaction.
  message Loaded {
    // Index of the Open binding for which this document was loaded.
    uint32 binding = 1;
    // Loaded JSON document.
    bytes doc_json = 2 [
        (gogoproto.casttype) = "encoding/json.RawMessage",
        json_name = "doc"
    ];
  }
  Loaded loaded = 5;

  // Flushed responds to a Request.Flush.
  // The driver will send no further Loaded responses.
  message Flushed {
    // Optional update to ConnectorState.
    // This update is durably written before the connector receives a following
    // Store or StartCommit request.
    flow.ConnectorState state = 1;
  }
  Flushed flushed = 6;

  // StartedCommit responds to a Request.StartCommit.
  // The driver has processed all Store requests, it has started to commit its
  // transaction (if it has one), and it is now ready for the runtime to start
  // committing to its own recovery log.
  message StartedCommit {
    // Optional *transactional* update to ConnectorState.
    // This update commits atomically with the Flow recovery log checkpoint.
    flow.ConnectorState state = 1;
  }
  StartedCommit started_commit = 7;

  // Notify the runtime that the previous transaction has committed.
  // On receipt, the runtime may begin to flush, store, and commit a
  // next (pipelined) transaction.
  //
  // Acknowledged is _not_ a direct response to Request.Acknowledge,
  // and Acknowledge vs Acknowledged may be written in either order.
  message Acknowledged {
    // Optional *non-transactional* update to ConnectorState.
    // This update is not transactional and the connector must tolerate a future,
    // duplicate Request.Acknowledge of this same checkpoint and connector state,
    // even after having previously responded with Acknowledged and a (discarded)
    // connector state update.
    flow.ConnectorState state = 1;
  }
  Acknowledged acknowledged = 8;

  // Reserved for internal use.
  bytes internal = 100 [ json_name = "$internal" ];
}

// Extra messages used by connectors
// TODO(johnny): Do we still need this?
message Extra {
  message ValidateExistingProjectionRequest {
    flow.MaterializationSpec.Binding existing_binding = 1;

    Request.Validate.Binding proposed_binding = 2;
  }

  message ValidateBindingAgainstConstraints {
    flow.MaterializationSpec.Binding binding = 1;

    map<string, Response.Validated.Constraint> constraints = 2;
  }
}