package main

import (
	"github.com/stripe/stripe-go/v72"
	"github.com/stripe/stripe-go/v72/sub"
)

//<SNIP>

func translateStatus(subStatus stripe.SubscriptionStatus) string {
	if subStatus == "past_due" {
		return "Past Due"
	} else {
		return string(subStatus)
	}
}

func (bsf BasicStripeFetcher) fetchSubscriptionStatus(subID string) (string, error) {
	stripe.Key = bsf.apiKey

	subscription, err := sub.Get(subID, nil)
	if err != nil {
		return "", err
	}

	status := translateStatus(subscription.Status)
	return status, nil
}

package main

import (
	"errors"
	"fmt"
	"log"

	"github.com/simpleforce/simpleforce"
)

type ProductData struct {
	accountId            string
	email                string
	givenName            string
	familyName           string
	planName             string
	stripeSubscriptionId string
	subscriptionStatus   string
	accountCreatedAt     string
}

type SalesforceUpdater interface {
	updateProductInstance(data ProductData) error
}

type BasicSalesforceUpdater struct {
	client *simpleforce.Client
}

//<SNIP>

func (bsu BasicSalesforceUpdater) query(data ProductData) error {
	q := fmt.Sprintf("SELECT FIELDS(ALL) FROM Product_Instance__c WHERE Workspace_Id__c = '%s' LIMIT 1", data.accountId)
	result, err := bsu.client.Query(q)
	if err != nil {
		return err
	}

	if len(result.Records) != 1 {
		return errors.New("unexpected query result")
	}

	obj := result.Records[0]

	if obj == nil {
		return errors.New("No Product Instance Found")
	}

	firstName := obj.StringField("Admin_First_Name__c")

	if firstName == "" {
		return errors.New("Couldn't fetch first name")
	}

	return nil
}

func (bsu BasicSalesforceUpdater) updateProductInstance(data ProductData) error {
	obj := bsu.client.SObject("Product_Instance__c").
		Set("ExternalIDField", "Workspace_Id__c").
		Set("Workspace_Id__c", data.accountId).
		Set("Name", "Org: "+data.accountId).
		Set("Admin_Email__c", data.email).
		Set("Admin_First_Name__c", data.givenName).
		Set("Admin_Last_Name__c", data.familyName).
		Set("Product__c", data.planName).
		Set("Stripe_Subscription_Id__c", data.stripeSubscriptionId).
		Set("Subscription_Status__c", data.subscriptionStatus).
		Set("Workspace_Created_At__c", data.accountCreatedAt).
		Upsert()

	if obj == nil {
		return errors.New("upsert failed!")
	}

	return nil
}

package main

//<SNIP>

func (a App) Run(v turbine.Turbine) error {
	platformDB, err := v.Resources("MY_DATA_WAREHOUSE")
	if err != nil {
		return err
	}

	configs := turbine.ResourceConfigs{
		turbine.ResourceConfig{
			Field: "table.types",
			Value: "VIEW",
		},
		turbine.ResourceConfig{
			Field: "incrementing.column.name",
			Value: "account_id",
		},
		turbine.ResourceConfig{
			Field: "validate.non.null",
			Value: "false",
		},
	}
	rr, err := platformDB.Records("tablename", configs)
	if err != nil {
		return err
	}

	//<SNIP>

	v.Process(rr, WriteToSalesforce{})

	return nil
}

//Converting the Turbine Record data to a form that's ready for
//sending to salesforce
func RecordToProductData(r turbine.Record) ProductData {
	accountId := r.Payload.Get("account_id").(float64)
	createdAt := r.Payload.Get("account_created_at").(float64)

	givenName, ok := r.Payload.Get("user_given_name").(string)
	if !ok {
		givenName = ""
	}

	familyName, ok := r.Payload.Get("user_family_name").(string)
	if !ok {
		familyName = ""
	}

	planName, ok := r.Payload.Get("plan_name").(string)
	if !ok {
		planName = ""
	}

	return ProductData{
		accountId:            strconv.Itoa(int(accountId)),
		email:                r.Payload.Get("user_email").(string),
		givenName:            givenName,
		familyName:           familyName,
		planName:             planName,
		stripeSubscriptionId: r.Payload.Get("stripe_subscription_id").(string),
		accountCreatedAt:     strconv.Itoa(int(createdAt)),
	}
}

type WriteToSalesforce struct{}

func (f WriteToSalesforce) Process(rr []turbine.Record) []turbine.Record {

  //<SNIP> fetching of env vars

	salesforceUpdater, err := newBasicSalesforceUpdater(salesforceInstanceUrl, salesforceClientId, salesforceUser, salesforcePassword, salesforceToken)
	if err != nil {
		log.Fatal("ERROR: salesforce updater creation failed")
	}

	for _, r := range rr {
		pd := RecordToProductData(r)
		subscriptionId := r.Payload.Get("stripe_subscription_id").(string)
		subscriptionStatus, err := subscriptionFetcher.fetchSubscriptionStatus(subscriptionId)

		if err != nil {
			continue
		}

	  //update our data with information from Stripe
		pd.subscriptionStatus = subscriptionStatus
		err = salesforceUpdater.updateProductInstance(pd)
	}

	// return original records unmodified
	return rr
}