// Get Message Function
// Takes SQS Object, Queue_url, and Message Attributes
// Returns a ReceiveMessageOutput Struct and Errors
func getMessage(queue *sqs.SQS, queue_url *string, message_attributes []*string) (sqs.Message, error) {

	// Params object of ReceiveMessageInput Struct
	params := &sqs.ReceiveMessageInput{
		QueueURL:              queue_url,
		MessageAttributeNames: message_attributes,
		MaxNumberOfMessages:   aws.Long(1),
		VisibilityTimeout:     aws.Long(1),
		WaitTimeSeconds:       aws.Long(1),
	}
	resp, err := queue.ReceiveMessage(params)

	message := *resp.Messages[0]
	return message, err
}

func createQueue(svc *sqs.SQS, queueName string) {
	params := &sqs.CreateQueueInput{
		QueueName: &queueName,
	}
	_, err := svc.CreateQueue(params)
	if err != nil {
		log.Fatal(err)
	}
}

func checkQueueSize(s *sqs.SQS, queueURL *string, avail, inFlight, delayed int64) {
	// Get All Attributes.
	respAttr, err := s.GetQueueAttributes(&sqs.GetQueueAttributesInput{
		QueueUrl:       queueURL,
		AttributeNames: []*string{aws.String(gqa.AttrAll)},
	})

	So(err, ShouldBeNil)
	So(*respAttr.Attributes[gqa.AttrApproximateNumberOfMessages], ShouldEqual, strconv.FormatInt(avail, 10))
	So(*respAttr.Attributes[gqa.AttrApproximateNumberOfMessagesNotVisible], ShouldEqual, strconv.FormatInt(inFlight, 10))
	So(*respAttr.Attributes[gqa.AttrApproximateNumberOfMessagesDelayed], ShouldEqual, strconv.FormatInt(delayed, 10))
}

func deleteMessage(svc *sqs.SQS, msgs ...*sqs.Message) error {
	// delete messages
	delParams := sqs.DeleteMessageInput{
		QueueUrl: queueURL,
	}

	for i, msg := range msgs {
		delParams.ReceiptHandle = msg.ReceiptHandle
		_, err := svc.DeleteMessage(&delParams)
		if err != nil {
			return err
		}

		fmt.Printf("message deleted %v: %v\n", i, *msg.Body)
	}

	return nil
}

// sendMessage sends msgs to SQS.
func sendMessage(svc *sqs.SQS, msgs ...string) error {
	// create message send params
	sendParams := sqs.SendMessageInput{
		QueueUrl: queueURL,
	}

	// send message
	for i, msg := range msgs {
		// message body is a datetime string
		sendParams.MessageBody = aws.String(msg)
		_, err := svc.SendMessage(&sendParams)
		if err != nil {
			return err
		}

		fmt.Printf("message sent %v: %v\n", i, msg)
	}

	return nil
}

func SendMessageBatch(s *sqs.SQS, req *sqs.SendMessageBatchInput, id2index map[string]int) error {
	resp, err := s.SendMessageBatch(req)
	if err != nil {
		return err
	}
	return queue.NewBatchError(id2index, resp.Failed)
}

func receive(qUrl string, svc *sqs.SQS, rx chan msg.Raw) {
	for {
		param := &sqs.ReceiveMessageInput{
			QueueUrl:            aws.String(qUrl),
			MaxNumberOfMessages: aws.Int64(int64(MaxNumberOfMessages)),
			VisibilityTimeout:   aws.Int64(int64(VisibilityTimeout)),
			WaitTimeSeconds:     aws.Int64(int64(WaitTimeSeconds)),
		}

		r, err := svc.ReceiveMessage(param)
		if err != nil {
			log.Println("WARN - problem receiving messages from SQS, sleeping, continuing.")
			time.Sleep(retry)
			continue
		}

		if (len(r.Messages)) > 0 {
			for _, raw := range r.Messages {
				m := msg.Raw{
					Body:          *raw.Body,
					ReceiptHandle: *raw.ReceiptHandle,
				}
				rx <- m
			}
		}
	}
}

func receive_from_queue(svc *sqs.SQS, queue_url *string) (*string, *string, *string, error) {
    // get sqs message
    params := &sqs.ReceiveMessageInput{
        QueueUrl: queue_url,
        MaxNumberOfMessages: aws.Int64(1),
        VisibilityTimeout: aws.Int64(1),
    }
    resp, err := svc.ReceiveMessage(params)
    if err != nil { return nil, nil, nil, err }

    // error on empty queue
    if len(resp.Messages) == 0 { return nil, nil, nil, errors.New("queue is empty") }

    receipt_handle := resp.Messages[0].ReceiptHandle
    body := resp.Messages[0].Body

    // unmarshal sqs message body
    data := &SqsBody{}
    err = json.Unmarshal([]byte(*body), &data)
    if err != nil { return nil, nil, nil, err }

    bucket := &data.Records[0].S3.Bucket.Name
    key := &data.Records[0].S3.Object.Key
    return receipt_handle, bucket, key, nil
}

func sendMessage(sqsInstance *sqs.SQS, queueURL *string, message string) error {
	_, err := sqsInstance.SendMessage(&sqs.SendMessageInput{
		QueueURL:    queueURL,
		MessageBody: &message,
	})
	return err
}

func Main() {
	var (
		SQS                        *sqs.SQS
		getUserQueueUrlOutput      *sqs.GetQueueUrlOutput
		getContainerQueueUrlOutput *sqs.GetQueueUrlOutput
		UserQueueUrl               *string
		ContainerQueueUrl          *string

		Dynamo *dynamodb.DynamoDB

		socialWorker *workers.SocialWorker
	)

	SQS = sqs.New(&aws.Config{Region: aws.String("cn-north-1")})
	getUserQueueUrlOutput, err := SQS.GetQueueUrl(&sqs.GetQueueUrlInput{QueueName: aws.String(USER_QUEUE_NAME)})
	if err != nil {
		glog.Errorln("Error on connect user queue url:", err.Error())
		return
	}
	UserQueueUrl = getUserQueueUrlOutput.QueueUrl
	getContainerQueueUrlOutput, err = SQS.GetQueueUrl(&sqs.GetQueueUrlInput{QueueName: aws.String(CONTAINER_QUEUE_NAME)})
	if err != nil {
		glog.Errorln("Error on connect container queue url:", err.Error())
		return
	}
	ContainerQueueUrl = getContainerQueueUrlOutput.QueueUrl

	Dynamo = dynamodb.New(&aws.Config{Region: aws.String("cn-north-1")})

	socialWorker = workers.NewSocialWorker(SQS, UserQueueUrl, ContainerQueueUrl, Dynamo)
	socialWorker.Start()
}

func get_queue(svc *sqs.SQS, queue string) (*string, error) {
    params := &sqs.GetQueueUrlInput{
        QueueName: aws.String(queue),
    }
    resp, err := svc.GetQueueUrl(params)
    if err != nil { return nil, err }
    return resp.QueueUrl, nil
}

func deleteQueue(s *sqs.SQS, queueName string) {
	resp, err := s.GetQueueUrl(&sqs.GetQueueUrlInput{QueueName: aws.String(queueName)})
	if err != nil {
		So(err.Error(), ShouldContainSubstring, "AWS.SimpleQueueService.NonExistentQueue")
	} else {
		_, e := s.DeleteQueue(&sqs.DeleteQueueInput{QueueUrl: resp.QueueUrl})
		So(e, ShouldBeNil)
	}
}

func delete_message(svc *sqs.SQS, queue_url *string, receipt_handle *string) error {
    params := &sqs.DeleteMessageInput{
        QueueUrl: queue_url,
        ReceiptHandle: receipt_handle,
    }
    _, err := svc.DeleteMessage(params)
    if err != nil { return err }
    return nil
}

func createQueue(sqsInstance *sqs.SQS, queueName string) (*string, error) {
	createQueueOutput, err := sqsInstance.CreateQueue(&sqs.CreateQueueInput{QueueName: &queueName})

	if err != nil {
		return nil, err
	}

	return createQueueOutput.QueueURL, nil
}

func getQueue(sqsInstance *sqs.SQS, queueName string) (*string, error) {
	queueURLOutput, err := sqsInstance.GetQueueURL(&sqs.GetQueueURLInput{QueueName: &queueName})

	if err != nil {
		return nil, err
	}

	return queueURLOutput.QueueURL, nil
}

func deleteMessage(svc *sqs.SQS, queueURL string, receiptHandle string) error {
	params := &sqs.DeleteMessageInput{
		QueueUrl:      aws.String(queueURL),      // Required
		ReceiptHandle: aws.String(receiptHandle), // Required
	}
	_, err := svc.DeleteMessage(params)
	if err != nil {
		return err
	}
	return nil
}

func createQueue(svc *sqs.SQS, queueName string) (string, error) {
	params := &sqs.CreateQueueInput{
		QueueName: aws.String(queueName), // Required
	}
	resp, err := svc.CreateQueue(params)

	if err != nil {
		return "", err
	}
	return *resp.QueueUrl, nil
}

func run(svc *sqs.SQS, params *sqs.ReceiveMessageInput) {
	for {
		resp, err := svc.ReceiveMessage(params)
		if err != nil {
			fmt.Println("There was an issue receiving a message: ", err)
		}

		renderIntoTemplate()

		if len(resp.Messages) > 0 {
			fmt.Printf("\nDisplay received message:\n\n%+v", resp)
		}

		time.Sleep(5 * time.Second) // message needs to be explicitly deleted still
	}
}

func delete(qUrl string, svc *sqs.SQS, dx chan string) {
	for {
		m := <-dx

		params := &sqs.DeleteMessageInput{
			QueueUrl:      aws.String(qUrl), // Required
			ReceiptHandle: aws.String(m),    // Required
		}

		_, err := svc.DeleteMessage(params)

		if err != nil {
			log.Println("WARN - problem deleting messages from SQS, continuing.")
		}

	}
}

func findQueueURL(sqsService *sqs.SQS, queueName string) string {
	// check the environment variable first
	var queueURL string
	if queueURL = os.Getenv("STREAMMARKER_SQS_QUEUE_URL"); queueURL != "" {
		return queueURL
	}

	// otherwise, query SQS for the queue URL
	params := &sqs.GetQueueUrlInput{
		QueueName: aws.String(queueName),
	}
	if resp, err := sqsService.GetQueueUrl(params); err == nil {
		queueURL = *resp.QueueUrl
	} else {
		stdlog.Panicf("Unable to retrieve queue URL: %s", err.Error())
	}
	return queueURL
}