"""Subscribe and Read Salesforce CDC events using Pub/Sub API
    Docs: https://developer.salesforce.com/docs/platform/pub-sub-api/guide/qs-python-quick-start.html
    Pub/Sub API is end of life in: June 30, 2025
"""
import datetime
import os

import grpc
import requests
import threading
import io
import pubsub_api_pb2 as pb2
import pubsub_api_pb2_grpc as pb2_grpc
import avro.schema
import avro.io
import certifi
import json
import xml.etree.ElementTree as ET


semaphore = threading.Semaphore(1)
latest_replay_id = None


def main():
    with open(certifi.where(), 'rb') as f:
        creds = grpc.ssl_channel_credentials(f.read())
    with grpc.secure_channel('api.pubsub.salesforce.com:7443', creds) as channel:

        # to login using username/password
        if not os.environ.get('sessionId'):
            details = login()
            print("\n login details = ", details, flush=True)
            os.environ["sessionId"] = details.get('sessionId')

        # actual run
        sessionId = os.environ["sessionId"]
        metadataServerUrl = 'https://sony--fullcopysb.sandbox.my.salesforce.com'
        organizationId = '00D7a0000005YxfEAE'
        authmetadata = (
            ('accesstoken', sessionId),
            ('instanceurl', metadataServerUrl),
            ('tenantid', organizationId)
        )
        stub = pb2_grpc.PubSubStub(channel)

        # Pre: enable this Object in CDC from UI. Ref: https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_select_objects.htm
        # Use this: to subscribe to all events by providing the standard ChangeEvents channel
        # mysubtopic = "/data/ChangeEvents"
        # Use this: to subscribe to fans change events only
        mysubtopic = "/data/Fan__ChangeEvent"
        print("\n Subscribing to " + mysubtopic, flush=True)
        substream = stub.Subscribe(
            fetchReqStream(mysubtopic),
           metadata=authmetadata
        )
        for event in substream:
            print('event==', event)
            if event.events:
                semaphore.release()
                print("\n Number of events received: ", len(event.events), flush=True)
                payloadbytes = event.events[0].event.payload
                schemaid = event.events[0].event.schema_id
                schema = stub.GetSchema(pb2.SchemaRequest(schema_id=schemaid),
                    metadata=authmetadata).schema_json
                decoded = decode(schema, payloadbytes)
                print("\n Got an event!", json.dumps(decoded), flush=True)
            else:
                print("\n [", datetime.datetime.now(), "] The subscription is active.", flush=True)
            latest_replay_id = event.latest_replay_id
            print('latest_replay_id==', latest_replay_id, flush=True)


def decode(schema, payload):
    schema = avro.schema.parse(schema)
    buf = io.BytesIO(payload)
    decoder = avro.io.BinaryDecoder(buf)
    reader = avro.io.DatumReader(schema)
    ret = reader.read(decoder)
    return ret


def fetchReqStream(topic):
    while True:
        semaphore.acquire()
        yield pb2.FetchRequest(
            topic_name=topic,
            replay_preset=pb2.ReplayPreset.LATEST,
            num_requested=1)


def parseXML(xmlfile, namespace):
    # create element tree object
    root = ET.fromstring(xmlfile)
    if not root or not root[0] or not root[0][0][0]:
        print('err res==', root[0][0])
        raise Exception('Invalid login response')

    data = {}
    for item in root[0][0][0]:
        if item.tag.replace(namespace, '') == 'userInfo':
            continue
        data[item.tag.replace(namespace, '')] = item.text.encode('utf8').decode() if item.text else ''
    return data


def login():
    # put
    username = 'kafkaintegration@sonymusic.com.fullcopysb'
    password = 'REPLACE-ME'
    url = 'https://sony--fullcopysb.sandbox.my.salesforce.com/services/Soap/u/59.0/'
    if not username or not password:
        raise Exception('salesforce credentials missing')

    headers = {'content-type': 'text/xml', 'SOAPAction': 'login'}
    xml = f"""<soapenv:Envelope xmlns:soapenv='http://schemas.xmlsoap.org/soap/envelope/'
    xmlns:xsi='http://www.w3.org/2001/XMLSchema-instance'
    xmlns:urn='urn:partner.soap.sforce.com'><soapenv:Body>
    <urn:login>
    <urn:username><![CDATA[{username}]]></urn:username>
    <urn:password><![CDATA[{password}]]></urn:password>
    </urn:login></soapenv:Body></soapenv:Envelope>"""
    res = requests.post(url, data=xml, headers=headers, verify=False)
    # Optionally, print the content field returned
    print('res==', res.status_code)
    if res.status_code > 200:
        print('**Error==', res.content)
        raise Exception('failed to login to salesforce')
    # print('content==', res.content)
    namespace = '{urn:partner.soap.sforce.com}'
    return parseXML(res.content, namespace)



# Press the green button in the gutter to run the script.
if __name__ == '__main__':
    # main()
    payload = b'\x00\x00\x00\x00\x00\x00\r\x16\x00\x00'
    buf = io.BytesIO(payload)
    print('buf==', buf)
    decoder = avro.io.BinaryDecoder(buf)
    print('decoded==', decoder)
    reader = avro.io.DatumReader()
    ret = reader.read(decoder)
    print('ret==', ret)