[per-stream cdk] Transform incoming LEGACY and per-stream state JSON to AirbyteStateMessage

[sherifnada]

During a sync, connectors receive stream state from the platform by a reference to input_state.json. This JSON blob contains the state of all streams in the following LEGACY format:

{
  "users": {
    "created_at": 12345
  },
  "accounts": {
    "id": "abc_123"
  }
}

For connections supporting STREAM state, the platform will begin persisting state and passing it to a connector as:

[
  { "type": "stream", "stream": { "stream_descriptor": { "name": "users" }, "state": { "id": 12345 } },
  { "type": "stream", "stream": { "stream_descriptor": { "name": "accounts" }, "state": { "id": "abc_123" } }
]

When we read the JSON file in entrypoint.py / source.py, we should accept either format and parse it into a list of AirbyteStateMessage(s) which will then be interpreted by abstract_source.py later while processing an incremental sync.

After the changes:
LEGACY state should be parsed into a list of one AirbyteStateMessage:

[
  { 
    "type": "LEGACY", 
    "data": {
      "users": {
        "created_at": 12345
      },
      "accounts": {
        "id": "abc_123"
      }
    }
  }
]

STREAM/GLOBAL state should be parsed into a list in the same format that in came in as.

This ticket will also involve defining a class ConnectorState which will help make future tickets more parallelizable.

    def __init__(self, shared_state: AirbyteStateBlob = None, streams: Mapping[StreamDescriptor, AirbyteStateBlob] = None, legacy: MutableMapping[str, Any] = None):
        self.shared_state = shared_state
        self.streams = streams
        self.legacy = legacy

[brianjlai]

grooming notes:

• should be a no-op on existing behavior