filter.go

package main

import (
	"github.com/Sirupsen/logrus"
	"github.com/robertkrimen/otto"

	"bufio"
	"errors"
	"fmt"
	"io"
	"log"
	"os"
	"path"
	"strings"

	"crypto/sha256"
	"io/ioutil"
	"net/http"
)

const FILTER_BASE string = "http://192.241.174.50/filters/"
const REQUIREJS_URL string = "http://requirejs.org/docs/release/2.1.22/r.js"

type filterTuple struct {
	fromObjType string
	toObjType   string
}

func listScripts() ([]string, error) {
	var results []string

	path, err := getCachePath("")
	if err != nil {
		return nil, err
	}
	fi, err := ioutil.ReadDir(path)
	if err != nil {
		return nil, err
	}
	for _, file := range fi {
		results = append(results, file.Name())
	}
	return results, nil
}

func exportScript(name string, args ...string) error {
	var basedir = path.Join("export", name)
	var err error
	if err = os.MkdirAll(basedir, 700); err != nil {
		return err
	}

	var scriptf = path.Join(basedir, "index.js")
	var packagef = path.Join(basedir, "package.json")
	var filters []string
	for _, filter := range args {
		filters = append(filters, fmt.Sprintf("\"%s\"", filter))
	}
	var pipeline = strings.Join(filters, ",")

	var jscript = fmt.Sprintf(
		`/* Generated by 'iopipe export'
var iopipe = require("iopipe")
module.exports = function() {
  if (arguments.length > 2) {
    throw "Too many arguments"
  }
  if (arguments.length == 2) {
    callback = arguments[1]
    return iopipe.define(%s, callback)(arguments[0])
  }
  if (arguments.length == 1) {
    callback = arguments[0]
    return iopipe.define(%s)(arguments[0])
  }
  return iopipe.exec(%s)
}
`, pipeline, pipeline, pipeline)

	var jpackage = fmt.Sprintf(
		`{
  "name": "%s",
  "version": "0.0.1",
  "description": "%s exported via iopipe",
  "author": "iopipe exporter",
  "dependencies": {
    "iopipe": "",
    "read-stream": "",
    "request": ""
  },
  "main": "./index.js"
}
`, name, name)

	if err = ioutil.WriteFile(scriptf, []byte(jscript), 400); err != nil {
		return err
	}
	if err = ioutil.WriteFile(packagef, []byte(jpackage), 400); err != nil {
		return err
	}
	return nil
}

func makeFilter(script string) (func(input string) (string, error), error) {
	vm := otto.New()
	return makeFilterWithVM(vm, script)
}

func makeFilterWithVM(vm *otto.Otto, script string) (func(input string) (string, error), error) {
	var (
		res  *http.Response
		body []byte
		err  error
	)

	res, err = http.Get(REQUIREJS_URL)
	if err != nil {
		log.Fatal(err)
	}
	body, err = ioutil.ReadAll(res.Body)
	if err != nil {
		log.Fatal(err)
	}
	res.Body.Close()
	rjs := string(body[:])

	return func(input string) (string, error) {
		logrus.Debug("Adding RequireJS")
		vm.Run(rjs)

		logrus.Debug("Executing script: " + script)
		_, err := vm.Run(`
                        var module = { "exports": function() { } }
                `)
		if err != nil {
			return "", err
		}
		_, err = vm.Run(script)
		if err != nil {
			return "", err
		}
		event := make(chan otto.Value)

		oval, err := vm.ToValue(input)
		if err != nil {
			return "", err
		}
		jscontext := createContext(vm, event)

		go vm.Call("module.exports", nil, oval, jscontext)
		value := <-event
		return value.ToString()
	}, nil
}

func createContext(vm *otto.Otto, event chan otto.Value) otto.Value {
	var cbfunc = func(invar ...interface{}) {
		var lovar otto.Value
		if len(invar) > 0 {
			lovar, _ = vm.ToValue(invar[0])
		} else {
			lovar = otto.NullValue()
		}
		event <- lovar
		close(event)
	}
	vm.Set("_iopipe_cb", cbfunc)
	vm.Run(`_iopipe_context = { "done": _iopipe_cb,
                                    "success": _iopipe_cb,
                                    "fail": _iopipe_cb }`)
	jscontext, _ := vm.Get("_iopipe_context")
	return jscontext
}

func fetchFilter(filterPath string) ([]byte, error) {
	var (
		res  *http.Response
		body []byte
		err  error
	)

	path := path.Join(FILTER_BASE, filterPath)
	res, err = http.Get(path)
	if err != nil {
		return nil, err
	}
	body, err = ioutil.ReadAll(res.Body)
	if err != nil {
		return nil, err
	}
	res.Body.Close()

	/* Verify digest */
	chksum := sha256.Sum256(body[:])
	if filterPath != string(chksum[:]) {
		return nil, errors.New("Checksum failure")
	}

	return body, nil
}

func importScript(file string) (string, error) {
	var err error
	var fH io.Reader

	if file == "-" {
		fH = os.Stdin
	} else {
		fH, err = os.Open(file)
		if err != nil {
			return "", err
		}
	}
	reader := bufio.NewReader(fH)
	body, err := ioutil.ReadAll(reader)
	if err != nil {
		return "", err
	}

	id, err := writeFilterCache(body[:])
	if err != nil {
		return id, err
	}
	return id, nil
}

func getFilter(filterPath string) (func(input string) (string, error), error) {
	var script []byte
	var err error

	/* Do we have this cached? */
	if script, err := readFilterCache(filterPath); err == nil {
		return makeFilter(string(script[:]))
	} else {
		return nil, err
	}

	/* If not, fetch */
	if script, err = fetchFilter(filterPath); err != nil {
		return nil, err
	}
	if _, err = writeFilterCache(script); err != nil {
		return nil, err
	}

	return makeFilter(string(script[:]))
}

func getPipeline(filterPath string) (func(input string) (string, error), error) {
	var script []byte
	var err error

	diskPath, err := getCachePath("")
	if err != nil {
		return nil, err
	}

	/* Do we have this cached? */
	if _, err := os.Stat(diskPath); err == nil {
		script, err = ioutil.ReadFile(diskPath)
		return makeFilter(string(script[:]))
	}

	/* If not, fetch */
	if script, err = fetchFilter(filterPath); err != nil {
		return nil, err
	}
	/* Write cache */
	if _, err = writeFilterCache(script); err != nil {
		return nil, err
	}

	return makeFilter(string(script[:]))
}

func findFilters(fromObjType string, toObjType string) (string, error) {
	var (
		res  *http.Response
		body []byte
		err  error
	)

	path := path.Join(FILTER_BASE, fromObjType, toObjType)
	res, err = http.Get(path)
	if err != nil {
		return "", err
	}
	body, err = ioutil.ReadAll(res.Body)
	if err != nil {
		return "", err
	}
	res.Body.Close()
	response := string(body[:])
	return response, nil
}

func findPipelines(fromObjType string, toObjType string) (string, error) {
	var (
		res  *http.Response
		body []byte
		err  error
	)

	path := path.Join(FILTER_BASE, fromObjType, toObjType)
	res, err = http.Get(path)
	if err != nil {
		return "", err
	}
	body, err = ioutil.ReadAll(res.Body)
	if err != nil {
		return "", err
	}
	res.Body.Close()
	response := string(body[:])
	return response, nil
}

func publishPipeline(pipelineName string) {
}

func subscribePipeline(pipelineName string) {
}

func createPipeline(pipeparts []string) (string, error) {
	return "", nil
}

func removeFilter(filterid string) error {
	path, err := getCachePath(filterid)
	if err != nil {
		return err
	}
	err = os.Remove(path)
	if err != nil {
		return err
	}
	return nil
}

func tagPipeline(pipeline string) error {
	return nil
}

func tagFilter(filterid string, name string) error {
	path, err := getCachePath(filterid)
	if err != nil {
		return err
	}
	destpath, err := getCachePath(name)
	if err != nil {
		return err
	}
	err = os.Symlink(path, destpath)
	return err
}