// NewPublisherImpl creates a Publisher that will submit certificates
// to any CT logs configured in CTConfig
func NewPublisherImpl(ctConfig CTConfig) (pub PublisherImpl, err error) {
	logger := blog.GetAuditLogger()
	logger.Notice("Publisher Authority Starting")

	if ctConfig.IntermediateBundleFilename == "" {
		err = fmt.Errorf("No CT submission bundle provided")
		return
	}
	bundle, err := core.LoadCertBundle(ctConfig.IntermediateBundleFilename)
	if err != nil {
		return
	}
	for _, cert := range bundle {
		pub.issuerBundle = append(pub.issuerBundle, base64.StdEncoding.EncodeToString(cert.Raw))
	}
	ctBackoff, err := time.ParseDuration(ctConfig.SubmissionBackoffString)
	if err != nil {
		return
	}
	for _, log := range ctConfig.Logs {
		if !strings.HasPrefix(log.URI, "https://") && !strings.HasPrefix(log.URI, "http://") {
			err = fmt.Errorf("Log URI [%s] is not absolute", log.URI)
			return
		}
	}

	pub.log = logger
	pub.client = &http.Client{}
	pub.submissionBackoff = ctBackoff
	pub.submissionRetries = ctConfig.SubmissionRetries
	pub.ctLogs = ctConfig.Logs

	return
}

// NewAmqpRPCServer creates a new RPC server for the given queue and will begin
// consuming requests from the queue. To start the server you must call Start().
func NewAmqpRPCServer(serverQueue string, maxConcurrentRPCServerRequests int64, c cmd.Config) (*AmqpRPCServer, error) {
	log := blog.GetAuditLogger()

	reconnectBase := c.AMQP.ReconnectTimeouts.Base.Duration
	if reconnectBase == 0 {
		reconnectBase = 20 * time.Millisecond
	}
	reconnectMax := c.AMQP.ReconnectTimeouts.Max.Duration
	if reconnectMax == 0 {
		reconnectMax = time.Minute
	}

	stats, err := statsd.NewClient(c.Statsd.Server, c.Statsd.Prefix)
	if err != nil {
		return nil, err
	}

	return &AmqpRPCServer{
		serverQueue:                    serverQueue,
		connection:                     newAMQPConnector(serverQueue, reconnectBase, reconnectMax),
		log:                            log,
		dispatchTable:                  make(map[string]func([]byte) ([]byte, error)),
		maxConcurrentRPCServerRequests: maxConcurrentRPCServerRequests,
		clk:   clock.Default(),
		stats: stats,
	}, nil
}

// NewDbMap creates the root gorp mapping object. Create one of these for each
// database schema you wish to map. Each DbMap contains a list of mapped tables.
// It automatically maps the tables for the primary parts of Boulder around the
// Storage Authority. This may require some further work when we use a disjoint
// schema, like that for `certificate-authority-data.go`.
func NewDbMap(driver string, dbConnect string) (dbmap *gorp.DbMap, err error) {
	logger := blog.GetAuditLogger()

	if driver == "mysql" {
		dbConnect, err = fixMysqlDSN(dbConnect)
		if err != nil {
			return
		}
	}

	db, err := sql.Open(driver, dbConnect)
	if err != nil {
		return
	}
	if err = db.Ping(); err != nil {
		return
	}

	logger.Debug(fmt.Sprintf("Connecting to database %s %s", driver, dbConnect))

	dialect, ok := dialectMap[driver].(gorp.Dialect)
	if !ok {
		err = fmt.Errorf("Couldn't find dialect for %s", driver)
		return
	}

	logger.Info(fmt.Sprintf("Connected to database %s %s", driver, dbConnect))

	dbmap = &gorp.DbMap{Db: db, Dialect: dialect, TypeConverter: BoulderTypeConverter{}}

	initTables(dbmap)

	return
}

// NewDbMap creates the root gorp mapping object. Create one of these for each
// database schema you wish to map. Each DbMap contains a list of mapped tables.
// It automatically maps the tables for the primary parts of Boulder around the
// Storage Authority. This may require some further work when we use a disjoint
// schema, like that for `certificate-authority-data.go`.
func NewDbMap(dbConnect string) (*gorp.DbMap, error) {
	logger := blog.GetAuditLogger()

	var err error
	dbConnect, err = recombineURLForDB(dbConnect)
	if err != nil {
		return nil, err
	}

	logger.Debug("Connecting to database")

	db, err := sql.Open("mysql", dbConnect)
	if err != nil {
		return nil, err
	}
	if err = db.Ping(); err != nil {
		return nil, err
	}

	dialect := gorp.MySQLDialect{Engine: "InnoDB", Encoding: "UTF8"}
	dbmap := &gorp.DbMap{Db: db, Dialect: dialect, TypeConverter: BoulderTypeConverter{}}

	initTables(dbmap)

	logger.Debug("Connected to database")

	return dbmap, err
}

func setup(t *testing.T) (OCSPUpdater, core.StorageAuthority, *gorp.DbMap, clock.FakeClock, func()) {
	dbMap, err := sa.NewDbMap(dbConnStr)
	test.AssertNotError(t, err, "Failed to create dbMap")

	fc := clock.NewFake()
	fc.Add(1 * time.Hour)

	sa, err := sa.NewSQLStorageAuthority(dbMap, fc)
	test.AssertNotError(t, err, "Failed to create SA")

	cleanUp := test.ResetSATestDatabase(t)

	stats, _ := statsd.NewNoopClient(nil)

	updater := OCSPUpdater{
		dbMap: dbMap,
		clk:   fc,
		cac:   &mockCA{},
		pubc:  &mockPub{sa},
		sac:   sa,
		stats: stats,
		log:   blog.GetAuditLogger(),
	}

	return updater, sa, dbMap, fc, cleanUp
}

func setup(t *testing.T, nagTimes []time.Duration) *testCtx {
	dbMap, err := sa.NewDbMap(dbConnStr)
	if err != nil {
		t.Fatalf("Couldn't connect the database: %s", err)
	}
	fc := clock.NewFake()
	ssa, err := sa.NewSQLStorageAuthority(dbMap, fc)
	if err != nil {
		t.Fatalf("unable to create SQLStorageAuthority: %s", err)
	}
	cleanUp := test.ResetTestDatabase(t, dbMap.Db)

	stats, _ := statsd.NewNoopClient(nil)
	mc := &mockMail{}

	m := &mailer{
		log:           blog.GetAuditLogger(),
		stats:         stats,
		mailer:        mc,
		emailTemplate: tmpl,
		dbMap:         dbMap,
		rs:            ssa,
		nagTimes:      nagTimes,
		limit:         100,
		clk:           fc,
	}
	return &testCtx{
		dbMap:   dbMap,
		ssa:     ssa,
		mc:      mc,
		fc:      fc,
		m:       m,
		cleanUp: cleanUp,
	}
}

func TestParseLine(t *testing.T) {
	dbMap, err := sa.NewDbMap(vars.DBConnSA)
	if err != nil {
		t.Fatalf("Failed to create dbMap: %s", err)
	}
	fc := clock.NewFake()
	fc.Set(time.Date(2015, 3, 4, 5, 0, 0, 0, time.UTC))
	sa, err := sa.NewSQLStorageAuthority(dbMap, fc)
	if err != nil {
		t.Fatalf("Failed to create SA: %s", err)
	}
	defer test.ResetSATestDatabase(t)()
	logger := blog.GetAuditLogger()

	found, added := parseLogLine(sa, logger, "")
	test.AssertEquals(t, found, false)
	test.AssertEquals(t, added, false)

	found, added = parseLogLine(sa, logger, "0000-00-00T00:00:00+00:00 hostname boulder-ca[pid]: [AUDIT] Failed RPC to store at SA, orphaning certificate: b64der=[] err=[AMQP-RPC timeout], regID=[1337]")
	test.AssertEquals(t, found, true)
	test.AssertEquals(t, added, false)

	found, added = parseLogLine(sa, logger, "0000-00-00T00:00:00+00:00 hostname boulder-ca[pid]: [AUDIT] Failed RPC to store at SA, orphaning certificate: b64der=[deadbeef] err=[AMQP-RPC timeout], regID=[]")
	test.AssertEquals(t, found, true)
	test.AssertEquals(t, added, false)

	reg := satest.CreateWorkingRegistration(t, sa)

	found, added = parseLogLine(sa, logger, fmt.Sprintf("0000-00-00T00:00:00+00:00 hostname boulder-ca[pid]: [AUDIT] Failed RPC to store at SA, orphaning certificate: b64der=[MIIEWzCCA0OgAwIBAgITAP+gFgYw1hiy61wFEIJLFCdIVjANBgkqhkiG9w0BAQsFADAfMR0wGwYDVQQDDBRoYXBweSBoYWNrZXIgZmFrZSBDQTAeFw0xNTEwMDMwNTIxMDBaFw0xNjAxMDEwNTIxMDBaMBgxFjAUBgNVBAMTDWV4YW1wbGUuY28uYm4wggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQCeo/HSH63lWW42pqdwlalHWOS3JGa3REraT3xM9v3psdRwuTtlwf3YlpF/JIzK5JtXyA3CHGSwEGmUMhMNBZ0tg5I0booXnHyUeDVUnGSnpWgMUY+vCly+pI5oT8pjBHdcj6kjnDTx1cstBjsJi9HBcYPHUh78iEZBsvC0FAKsh8cHaEjUNHzvWd1anBdK0lRn25M8le9IxXi6di9SeyFmahmPteH+LYKZtNzrF5HpatB14+ywV8d212T62PCCnUPDLd+YWjo2+t5pZs7IlGhyGh7EerOOrI2kUUBg3tUdKDp4e3xplxvaAfSfdrqkGx+bQ0iqQnng+lVkXWYWRB8NAgMBAAGjggGVMIIBkTAOBgNVHQ8BAf8EBAMCBaAwHQYDVR0lBBYwFAYIKwYBBQUHAwEGCCsGAQUFBwMCMAwGA1UdEwEB/wQCMAAwHQYDVR0OBBYEFDadDBAEUrnrP/566FLp6DmjrlrbMB8GA1UdIwQYMBaAFPt4TxL5YBWDLJ8XfzQZsy426kGJMGoGCCsGAQUFBwEBBF4wXDAmBggrBgEFBQcwAYYaaHR0cDovL2xvY2FsaG9zdDo0MDAyL29jc3AwMgYIKwYBBQUHMAKGJmh0dHA6Ly9sb2NhbGhvc3Q6NDAwMC9hY21lL2lzc3Vlci1jZXJ0MBgGA1UdEQQRMA+CDWV4YW1wbGUuY28uYm4wJwYDVR0fBCAwHjAcoBqgGIYWaHR0cDovL2V4YW1wbGUuY29tL2NybDBjBgNVHSAEXDBaMAoGBmeBDAECATAAMEwGAyoDBDBFMCIGCCsGAQUFBwIBFhZodHRwOi8vZXhhbXBsZS5jb20vY3BzMB8GCCsGAQUFBwICMBMMEURvIFdoYXQgVGhvdSBXaWx0MA0GCSqGSIb3DQEBCwUAA4IBAQC7tLmUlxyvouVuIljbRtiL+zYdi/zXVSHAMXTkceqp8/8ucZBZu1fMBkB5SW2FUFd8EnuqhKGOeS3dNr9Pe4dLbUDR0UKIwV045Na+Jet4BbHDdWs3NXAutFhdGIa8ivLBQIbTzlBuVRhJE8g6qqjf5hYL0DXkLNptl2l+0+4xJMm/liCp/mYCGRwbdGUzwdSjACO76QLLSqZhkBF37ZJOuDbJTMBi3QzkOcTs6e4d/gSZpCy7yy6nJDxZ9N9P3XBYIpus+aZAYy29d2shYzE3st8cQfB2Wmb0SHd67sftTAzeudiiNW/4E4IKKH4R1S794apUO07y7pkqep1cz32k] err=[AMQP-RPC timeout], regID=[%d]", reg.ID))
	test.AssertEquals(t, found, true)
	test.AssertEquals(t, added, true)
}

// Response is called by the HTTP server to handle a new OCSP request.
func (src *DBSource) Response(req *ocsp.Request) (response []byte, present bool) {
	log := blog.GetAuditLogger()

	// Check that this request is for the proper CA
	if bytes.Compare(req.IssuerKeyHash, src.caKeyHash) != 0 {
		log.Debug(fmt.Sprintf("Request intended for CA Cert ID: %s", hex.EncodeToString(req.IssuerKeyHash)))
		present = false
		return
	}

	serialString := core.SerialToString(req.SerialNumber)
	log.Debug(fmt.Sprintf("Searching for OCSP issued by us for serial %s", serialString))

	var ocspResponse core.OCSPResponse
	// Note: we order by id rather than createdAt, because otherwise we sometimes
	// get the wrong result if a certificate is revoked in the same second as its
	// last update (e.g. client issues and instant revokes).
	err := src.dbMap.SelectOne(&ocspResponse, "SELECT * from ocspResponses WHERE serial = :serial ORDER BY id DESC LIMIT 1;",
		map[string]interface{}{"serial": serialString})
	if err != nil {
		present = false
		return
	}

	log.Info(fmt.Sprintf("OCSP Response sent for CA=%s, Serial=%s", hex.EncodeToString(src.caKeyHash), serialString))

	response = ocspResponse.Response
	present = true
	return
}

func NewAmqpRPCCLient(clientQueue, serverQueue string, channel *amqp.Channel) (rpc *AmqpRPCCLient, err error) {
	rpc = &AmqpRPCCLient{
		serverQueue: serverQueue,
		clientQueue: clientQueue,
		channel:     channel,
		pending:     make(map[string]chan []byte),
		timeout:     10 * time.Second,
		log:         blog.GetAuditLogger(),
	}

	// Subscribe to the response queue and dispatch
	msgs, err := amqpSubscribe(rpc.channel, clientQueue, nil)
	if err != nil {
		return
	}

	go func() {
		for msg := range msgs {
			// XXX-JWS: jws.Sign(privKey, body)
			corrID := msg.CorrelationId
			responseChan, present := rpc.pending[corrID]

			rpc.log.Debug(fmt.Sprintf(" [c<] received %s(%s) [%s]", msg.Type, core.B64enc(msg.Body), corrID))
			if !present {
				continue
			}
			responseChan <- msg.Body
			delete(rpc.pending, corrID)
		}
	}()

	return
}

// NewDbMap creates the root gorp mapping object. Create one of these for each
// database schema you wish to map. Each DbMap contains a list of mapped tables.
// It automatically maps the tables for the primary parts of Boulder around the
// Storage Authority. This may require some further work when we use a disjoint
// schema, like that for `certificate-authority-data.go`.
func NewDbMap(driver string, name string) (*gorp.DbMap, error) {
	logger := blog.GetAuditLogger()

	// We require this parameter for MySQL, so fail now if it is not present
	if driver == "mysql" && !strings.Contains(name, "parseTime=true") {
		return nil, fmt.Errorf("Database name must have parseTime=true")
	}

	db, err := sql.Open(driver, name)
	if err != nil {
		return nil, err
	}
	if err = db.Ping(); err != nil {
		return nil, err
	}

	logger.Debug(fmt.Sprintf("Connecting to database %s %s", driver, name))

	dialect, ok := dialectMap[driver].(gorp.Dialect)
	if !ok {
		err = fmt.Errorf("Couldn't find dialect for %s", driver)
		return nil, err
	}

	logger.Info(fmt.Sprintf("Connected to database %s %s", driver, name))

	dbmap := &gorp.DbMap{Db: db, Dialect: dialect, TypeConverter: BoulderTypeConverter{}}

	initTables(dbmap)

	return dbmap, err
}

// SetSQLDebug enables/disables GORP SQL-level Debugging
func SetSQLDebug(dbMap *gorp.DbMap, state bool) {
	dbMap.TraceOff()

	if state {
		// Enable logging
		dbMap.TraceOn("SQL: ", &SQLLogger{blog.GetAuditLogger()})
	}
}

// NewRegistrationAuthorityImpl constructs a new RA object.
func NewRegistrationAuthorityImpl() RegistrationAuthorityImpl {
	logger := blog.GetAuditLogger()
	logger.Notice("Registration Authority Starting")

	ra := RegistrationAuthorityImpl{log: logger}
	ra.PA = policy.NewPolicyAuthorityImpl()
	return ra
}

// Create a new AMQP-RPC server on the given queue and channel.
// Note that you must call Start() to actually start the server
// listening for requests.
func NewAmqpRPCServer(serverQueue string, channel *amqp.Channel) *AmqpRPCServer {
	log := blog.GetAuditLogger()
	return &AmqpRPCServer{
		serverQueue:   serverQueue,
		channel:       channel,
		log:           log,
		dispatchTable: make(map[string]func([]byte) []byte),
	}
}

// FailOnError exits and prints an error message if we encountered a problem
func FailOnError(err error, msg string) {
	if err != nil {
		// AUDIT[ Error Conditions ] 9cc4d537-8534-4970-8665-4b382abe82f3
		logger := blog.GetAuditLogger()
		logger.Err(fmt.Sprintf("%s: %s", msg, err))
		fmt.Fprintf(os.Stderr, "%s: %s\n", msg, err)
		os.Exit(1)
	}
}

// NewPublisherImpl creates a Publisher that will submit certificates
// to any CT logs configured in CTConfig
func NewPublisherImpl(bundle []ct.ASN1Cert, logs []*Log) (pub PublisherImpl) {
	logger := blog.GetAuditLogger()
	logger.Notice("Publisher Authority Starting")

	pub.issuerBundle = bundle
	pub.log = logger
	pub.ctLogs = logs

	return
}

// RandomString returns a randomly generated string of the requested length.
func RandomString(byteLength int) string {
	b := make([]byte, byteLength)
	_, err := io.ReadFull(rand.Reader, b)
	if err != nil {
		ohdear := "RandomString entropy failure? " + err.Error()
		logger := blog.GetAuditLogger()
		logger.EmergencyExit(ohdear)
	}
	return B64enc(b)
}

// NewCertificateAuthorityDatabaseImpl constructs a Database for the
// Certificate Authority.
func NewCertificateAuthorityDatabaseImpl(dbMap *gorp.DbMap) (cadb core.CertificateAuthorityDatabase, err error) {
	logger := blog.GetAuditLogger()

	dbMap.AddTableWithName(SerialNumber{}, "serialNumber").SetKeys(true, "ID")

	cadb = &CertificateAuthorityDatabaseImpl{
		dbMap: dbMap,
		log:   logger,
	}
	return cadb, nil
}

// NewSQLStorageAuthority provides persistence using a SQL backend for
// Boulder. It will modify the given gorp.DbMap by adding relevent tables.
func NewSQLStorageAuthority(dbMap *gorp.DbMap) (*SQLStorageAuthority, error) {
	logger := blog.GetAuditLogger()

	logger.Notice("Storage Authority Starting")

	ssa := &SQLStorageAuthority{
		dbMap: dbMap,
		log:   logger,
	}

	return ssa, nil
}

// NewPolicyAuthorityImpl constructs a Policy Authority.
func NewPolicyAuthorityImpl() *PolicyAuthorityImpl {
	logger := blog.GetAuditLogger()
	logger.Notice("Policy Authority Starting")

	pa := PolicyAuthorityImpl{log: logger}

	// TODO: Add configurability
	pa.PublicSuffixList = PublicSuffixList
	pa.Blacklist = blacklist

	return &pa
}

// NewAmqpRPCClient constructs an RPC client using AMQP
func NewAmqpRPCClient(clientQueuePrefix, serverQueue string, channel *amqp.Channel, stats statsd.Statter) (rpc *AmqpRPCCLient, err error) {
	hostname, err := os.Hostname()
	if err != nil {
		return nil, err
	}

	randID := make([]byte, 3)
	_, err = rand.Read(randID)
	if err != nil {
		return nil, err
	}
	clientQueue := fmt.Sprintf("%s.%s.%x", clientQueuePrefix, hostname, randID)

	rpc = &AmqpRPCCLient{
		serverQueue: serverQueue,
		clientQueue: clientQueue,
		channel:     channel,
		pending:     make(map[string]chan []byte),
		timeout:     10 * time.Second,
		log:         blog.GetAuditLogger(),
		stats:       stats,
	}

	// Subscribe to the response queue and dispatch
	msgs, err := amqpSubscribe(rpc.channel, clientQueue, "", rpc.log)
	if err != nil {
		return nil, err
	}

	go func() {
		for msg := range msgs {
			// XXX-JWS: jws.Sign(privKey, body)
			corrID := msg.CorrelationId
			rpc.mu.RLock()
			responseChan, present := rpc.pending[corrID]
			rpc.mu.RUnlock()

			rpc.log.Debug(fmt.Sprintf(" [c<][%s] response %s(%s) [%s]", clientQueue, msg.Type, core.B64enc(msg.Body), corrID))
			if !present {
				// AUDIT[ Misrouted Messages ] f523f21f-12d2-4c31-b2eb-ee4b7d96d60e
				rpc.log.Audit(fmt.Sprintf(" [c<][%s] Misrouted message: %s - %s - %s", clientQueue, msg.Type, core.B64enc(msg.Body), msg.CorrelationId))
				continue
			}
			responseChan <- msg.Body
			rpc.mu.Lock()
			delete(rpc.pending, corrID)
			rpc.mu.Unlock()
		}
	}()

	return rpc, err
}